"""External admin domain - business logic.""" from __future__ import annotations import hashlib import logging import secrets import tarfile from datetime import datetime, timedelta, timezone from typing import Optional from urllib.parse import urlencode from uuid import UUID import httpx from src.config import get_settings from src.core.exceptions import ( ConflictError, ForbiddenError, NotFoundError, PaymentRequiredError, ValidationError, ) from src.core.security.encryption import encrypt_value from src.domain.billing.access_policy import BillingAccessPolicy from src.domain.external_admin.models import ( ACTION_SESSION_STATUS_APPROVED, ACTION_SESSION_STATUS_APPLIED, ACTION_SESSION_STATUS_EXECUTING, ACTION_SESSION_STATUS_FAILED, ACTION_SESSION_STATUS_REJECTED, CHAT_JOB_STATUS_PENDING, CREDENTIAL_STATUS_ACTIVE, CREDENTIAL_STATUS_ROTATED, INSTALLATION_STATUS_CONNECTED, WP_MUTATION_ACTION_STATUS_COMPLETED, WP_MUTATION_ACTION_STATUS_FAILED as WP_MUTATION_FAILED, WP_MUTATION_ACTION_STATUS_PENDING, WP_MUTATION_ACTION_STATUS_RUNNING, WpMutationAction, INSTALLATION_STATUS_DISCONNECTED, INSTALLATION_STATUS_PENDING, INSTALLATION_STATUS_REDACTED, INSTALLATION_STATUS_TAMPERED, INSTALLATION_TYPE_SHOPIFY, INSTALLATION_TYPE_WORDPRESS, SITE_UNDERSTANDING_STATUS_QUEUED, AssistantActionSession, AssistantDataConsent, AssistantProjectSettings, ExternalAdminChatJob, ExternalAdminChatMessage, ExternalAdminChatSession, ExternalAdminInstallation, ExternalCredential, SiteUnderstandingRun, ) from src.domain.external_admin.repository import ( AssistantActionSessionRepository, AssistantDataConsentRepository, AssistantProjectSettingsRepository, ExternalAdminChatJobRepository, ExternalAdminChatMessageRepository, ExternalAdminChatSessionRepository, ExternalAdminInstallationRepository, ExternalCredentialRepository, SiteUnderstandingRunRepository, WpMutationActionRepository, ) from src.domain.external_admin.schemas import ( AssistantProjectSettingsUpdate, CompanionInstallRequest, CompanionInstallResponse, ShopifyConnectProjectResponse, ShopifyApplyPollRequest, ShopifyApplyPollResponse, ShopifyApplyReportRequest, ShopifyApplyReportResponse, ShopifyChatJobPollRequest, ShopifyChatJobPollResponse, ShopifyChatMessageItem, ShopifyChatMessageListRequest, ShopifyChatMessageListResponse, ShopifyChatMessageSendRequest, ShopifyChatMessageSendResponse, ShopifyChatSessionOpenRequest, ShopifyChatSessionOpenResponse, ShopifyHeartbeatRequest, ShopifyHeartbeatResponse, ShopifyOAuthCallbackRequest, ShopifyWebhookEnvelope, ShopifySiteUnderstandingStatusRequest, ShopifySiteUnderstandingStatusResponse, ShopifySiteUnderstandingSyncRequest, ShopifySiteUnderstandingSyncResponse, WordPressConnectCodeResponse, WordPressConnectCompleteRequest, WordPressConnectCompleteResponse, WordPressConnectStartRequest, WordPressEmbedExchangeRequest, WordPressEmbedExchangeResponse, WordPressChatJobPollRequest, WordPressChatJobPollResponse, WordPressChatMessageItem, WordPressChatMessageListRequest, WordPressChatMessageListResponse, WordPressChatMessageSendRequest, WordPressChatMessageSendResponse, WordPressChatSessionOpenRequest, WordPressChatSessionOpenResponse, WordPressCompanionMutationPollRequest, WordPressCompanionMutationPollResponse, WordPressCompanionMutationReportRequest, WordPressCompanionMutationReportResponse, WordPressHeartbeatRequest, WordPressHeartbeatResponse, WordPressHostedUrlRequest, WordPressHostedUrlResponse, WordPressSiteUnderstandingStatusRequest, WordPressSiteUnderstandingStatusResponse, WordPressSiteUnderstandingSyncRequest, WordPressSiteUnderstandingSyncResponse, ) from src.domain.external_admin.wordpress_connect import ( _normalize_site_url, issue_wordpress_connect_code, redeem_wordpress_connect_code, validate_wordpress_return_url, resolve_wp_plugin_api_url, ) from src.domain.external_admin.wordpress_embed import ( issue_wordpress_embed_token, mint_embed_access_token, verify_wordpress_embed_token, ) from src.domain.external_admin.wordpress_companion_package import ( resolve_wordpress_companion_package, ) from src.domain.projects.repository import ProjectRepository # Heartbeat is considered stale after this many minutes without contact. HEARTBEAT_STALE_MINUTES = 30 # Cache of companion archive sha256 → expected dir hash so we don't re-extract on every heartbeat. _companion_hash_cache: dict[str, str] = {} def _hash_result(value: dict) -> str: import json canonical = json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")) return hashlib.sha256(canonical.encode()).hexdigest() def _compute_companion_expected_hash() -> str | None: """SHA-256 over sorted PHP files in the companion archive. Algorithm is identical to Heartbeat::dirHash() in the WP plugin: for each .php file in sorted(relative_path): sha256.update(relative_path + "\\n") sha256.update(file_bytes) """ package = resolve_wordpress_companion_package() if package is None: return None cache_key = package.sha256 if cache_key in _companion_hash_cache: return _companion_hash_cache[cache_key] files: dict[str, bytes] = {} try: with tarfile.open(package.path, "r:gz") as archive: for member in archive.getmembers(): if not member.isfile() or not member.name.endswith(".php"): continue from pathlib import Path as _Path path = _Path(member.name) parts = ( path.parts[1:] if path.parts and path.parts[0] == "P2HS-AI-ASSIST-WP-Companion" else path.parts ) if not parts or any(p in (".git", "node_modules", "vendor") for p in parts): continue extracted = archive.extractfile(member) if extracted is None: continue files["/".join(parts)] = extracted.read() except Exception: return None if not files: return None ctx = hashlib.sha256() for rel_path in sorted(files): ctx.update((rel_path + "\n").encode()) ctx.update(files[rel_path]) result = ctx.hexdigest() _companion_hash_cache[cache_key] = result return result _CHAT_PLACEHOLDER_CONTENT = ( "Core assistant orchestration is being connected. Your message has been received and " "will be routed through a Core assistant run with Site Understanding context once the " "AI pipeline is fully wired." ) _CHAT_SESSION_TTL_HOURS = 8 def _format_chat_message(row: ExternalAdminChatMessage) -> WordPressChatMessageItem: return WordPressChatMessageItem( id=str(row.id), role=row.role, content=row.content, created_at=row.created_at.replace(microsecond=0).isoformat(), ) def _format_shopify_chat_message(row: ExternalAdminChatMessage) -> ShopifyChatMessageItem: return ShopifyChatMessageItem( id=str(row.id), role=row.role, content=row.content, created_at=row.created_at.replace(microsecond=0).isoformat(), ) logger = logging.getLogger(__name__) # Shopify's mandatory privacy webhooks. Handled in _handle_shopify_privacy_webhook. SHOPIFY_PRIVACY_TOPICS = frozenset({"customers/data_request", "customers/redact", "shop/redact"}) class ExternalAdminService: def __init__( self, installation_repo: ExternalAdminInstallationRepository, credential_repo: ExternalCredentialRepository, settings_repo: AssistantProjectSettingsRepository, action_session_repo: AssistantActionSessionRepository, consent_repo: AssistantDataConsentRepository, project_repo: ProjectRepository, chat_session_repo: ExternalAdminChatSessionRepository, chat_message_repo: ExternalAdminChatMessageRepository, chat_job_repo: ExternalAdminChatJobRepository, site_understanding_run_repo: SiteUnderstandingRunRepository, mutation_action_repo: WpMutationActionRepository, billing_policy: Optional[BillingAccessPolicy] = None, ) -> None: self._installations = installation_repo self._credentials = credential_repo self._settings = settings_repo self._action_sessions = action_session_repo self._consents = consent_repo self._projects = project_repo self._chat_sessions = chat_session_repo self._chat_messages = chat_message_repo self._chat_jobs = chat_job_repo self._su_runs = site_understanding_run_repo self._mutation_actions = mutation_action_repo self._billing_policy = billing_policy # ── Installation ────────────────────────────────────────────────────────── async def get_installation(self, project_id: UUID, workspace_id: UUID) -> ExternalAdminInstallation: await self._assert_project_in_workspace(project_id, workspace_id) row = await self._installations.get_by_project(project_id) if not row: raise NotFoundError("external_admin_installation_not_found") return row # ── WordPress connect ───────────────────────────────────────────────────── async def handle_wordpress_connect_complete( self, body: WordPressConnectCompleteRequest, redis, ) -> WordPressConnectCompleteResponse: payload = await redeem_wordpress_connect_code(redis, body.connect_code, body.site_url) try: installation_uuid = UUID(str(payload["installation_id"])) except ValueError as exc: raise ValidationError("invalid_installation_id") from exc installation = await self._installations.get_by_id(installation_uuid) if not installation: raise NotFoundError("external_admin_installation_not_found") if installation.installation_type != INSTALLATION_TYPE_WORDPRESS: raise ForbiddenError("installation_type_mismatch") now = datetime.now(timezone.utc) # The credential is minted here, at redemption, and returned in the response body only. # It is never part of the code (which travels in a browser URL). raw_token = secrets.token_urlsafe(32) installation.installation_token_enc = self._encrypt(raw_token) installation.metadata_json = { **(installation.metadata_json or {}), "installation_token_hash": self._token_hash(raw_token), } installation.site_url = body.site_url installation.status = INSTALLATION_STATUS_CONNECTED installation.connected_at = installation.connected_at or now installation.disconnected_at = None await self._installations.save(installation) embed_url = self._wordpress_embed_page_url(installation, body.site_url) return WordPressConnectCompleteResponse( ok=True, installation_id=str(installation.id), project_id=str(installation.project_id), credential_token=raw_token, core_api_url=resolve_wp_plugin_api_url(), app_url=get_settings().app_url.rstrip("/"), embed_url=embed_url, ) async def build_wordpress_connect_start_url( self, body: WordPressConnectStartRequest, ) -> WordPressHostedUrlResponse: app_url = get_settings().app_url.rstrip("/") # Refuse to build a Dashboard link that would send the browser (and a connect code) # anywhere but back to this site's wp-admin. return_url = validate_wordpress_return_url(body.site_url, body.return_url) query = urlencode( { "type": "ai_assist_wordpress", "site_url": body.site_url, "return_url": return_url, } ) return WordPressHostedUrlResponse(ok=True, url=f"{app_url}/projects?{query}") async def find_existing_wordpress_project_id( self, workspace_id: UUID, site_url: str, ) -> Optional[UUID]: """Return the project_id of an existing WordPress installation for this site. Prevents duplicate projects when the same site reconnects. Matches on the normalized site_url within the workspace. """ normalized = _normalize_site_url(site_url) existing = await self._installations.get_by_workspace_and_site_url(workspace_id, normalized) return existing.project_id if existing else None async def issue_wordpress_connect_code( self, project_id: UUID, workspace_id: UUID, site_url: str, redis, return_url: Optional[str] = None, ) -> WordPressConnectCodeResponse: project = await self._projects.get_by_id(project_id) if not project or project.workspace_id != workspace_id: raise ForbiddenError("project_not_in_workspace") if project.type != "ai_assist_wordpress": raise ForbiddenError("project_type_mismatch") if return_url: # Server-side enforcement: the Dashboard checks this too, but a code must never be # issued for a return address that is not this site's wp-admin. validate_wordpress_return_url(site_url, return_url) site_url = _normalize_site_url(site_url) installation = await self._installations.get_by_project(project_id) if not installation: installation = ExternalAdminInstallation( project_id=project_id, workspace_id=workspace_id, installation_type=INSTALLATION_TYPE_WORDPRESS, status=INSTALLATION_STATUS_PENDING, site_url=site_url, ) await self._installations.create(installation) elif installation.installation_type != INSTALLATION_TYPE_WORDPRESS: raise ForbiddenError("installation_type_mismatch") else: installation.site_url = site_url await self._installations.save(installation) code = await issue_wordpress_connect_code( redis, installation_id=str(installation.id), site_url=site_url, ) return WordPressConnectCodeResponse( ok=True, installation_id=str(installation.id), connect_code=code, core_api_url=resolve_wp_plugin_api_url(), app_url=get_settings().app_url.rstrip("/"), ) async def build_wordpress_embed_url( self, bearer_token: str, body: WordPressHostedUrlRequest, ) -> WordPressHostedUrlResponse: installation = await self._resolve_wordpress_installation_for_hosted_url(bearer_token, body) return WordPressHostedUrlResponse( ok=True, url=self._wordpress_embed_page_url(installation, body.site_url), ) async def handle_wordpress_embed_exchange( self, body: WordPressEmbedExchangeRequest, ) -> WordPressEmbedExchangeResponse: payload = verify_wordpress_embed_token(body.embed_token, project_id=body.project_id) try: installation_uuid = UUID(payload["installation_id"]) project_uuid = UUID(payload["project_id"]) except ValueError as exc: raise ValidationError("invalid_embed_token_ids") from exc installation = await self._installations.get_by_id(installation_uuid) if not installation or installation.project_id != project_uuid: raise NotFoundError("external_admin_installation_not_found") project = await self._projects.get_by_id(project_uuid) if not project or not project.created_by: raise ForbiddenError("embed_project_owner_missing") access_token, expires_at = mint_embed_access_token(project.created_by) return WordPressEmbedExchangeResponse( ok=True, access_token=access_token, expires_at=expires_at, user_id=str(project.created_by), workspace_id=str(project.workspace_id), project_id=str(project.id), installation_id=str(installation.id), site_url=payload["site_url"], ) def _wordpress_embed_page_url(self, installation: ExternalAdminInstallation, site_url: str) -> str: app_url = get_settings().app_url.rstrip("/") token = issue_wordpress_embed_token( installation_id=str(installation.id), project_id=str(installation.project_id), workspace_id=str(installation.workspace_id), site_url=site_url, ) query = urlencode( { "embed_token": token, "installation_id": str(installation.id), "site_url": _normalize_site_url(site_url), } ) return f"{app_url}/embed/wordpress/{installation.project_id}?{query}" async def _wordpress_site_readiness( self, installation: ExternalAdminInstallation, ) -> dict: project = await self._projects.get_by_id(installation.project_id) latest_run = await self._su_runs.get_latest_for_installation(installation.id) latest_completed = await self._su_runs.get_latest_completed_for_installation( installation.id ) project_ready = bool(project and project.status == "ready") connection_ready = installation.status == INSTALLATION_STATUS_CONNECTED companion_ready = bool((installation.companion_version or "").strip()) sync_ready = bool(latest_completed and latest_completed.completed_at) billing_ready = True billing_message = "Billing access is active." billing_status = "unknown" if self._billing_policy: try: billing_status = await self._billing_policy.get_billing_status(installation.workspace_id) await self._billing_policy.require_active_billing(installation.workspace_id) except PaymentRequiredError as exc: billing_ready = False billing_status = getattr(exc, "billing_status", None) or billing_status billing_message = "Active billing or a valid free plan is required before assistant actions can run." sync_status = latest_run.status if latest_run else "not_started" sync_running = sync_status in {"queued", "running"} sync_failed = sync_status == "failed" steps = [ { "key": "connection", "label": "P2HS connection", "status": "complete" if connection_ready else "blocked", "required": True, "blocking": not connection_ready, "message": "This WordPress site is connected." if connection_ready else "Reconnect this WordPress site to P2HS.", }, { "key": "billing", "label": "P2HS billing", "status": "complete" if billing_ready else "blocked", "required": True, "blocking": not billing_ready, "message": billing_message, }, { "key": "project", "label": "P2HS project", "status": "complete" if project_ready else "pending", "required": True, "blocking": not project_ready, "message": "The P2HS project workspace is ready." if project_ready else "P2HS is still preparing the project workspace.", }, { "key": "companion", "label": "Companion plugin", "status": "complete" if companion_ready else "pending", "required": True, "blocking": not companion_ready, "message": "Companion plugin starter is installed." if companion_ready else "Install the P2HS companion plugin before assistant actions can run.", }, { "key": "site_understanding", "label": "First site sync", "status": ( "complete" if sync_ready else "running" if sync_running else "failed" if sync_failed else "pending" ), "required": True, "blocking": not sync_ready, "message": "Site Understanding has completed." if sync_ready else "Site Understanding is running." if sync_running else "The latest Site Understanding sync failed. Run sync again." if sync_failed else "Run the first Site Understanding sync before chatting.", }, ] blocking_reasons = [ step["key"] for step in steps if step["blocking"] ] ready = not blocking_reasons next_action = "ready" if not connection_ready: next_action = "reconnect" elif not billing_ready: next_action = "resolve_billing" elif not project_ready: next_action = "wait_for_project" elif not companion_ready: next_action = "install_companion" elif sync_running: next_action = "wait_for_sync" elif not sync_ready: next_action = "sync_site" return { "ready": ready, "status": "ready" if ready else "setup_required", "next_action": next_action, "blocking_reasons": blocking_reasons, "project_id": str(installation.project_id), "workspace_id": str(installation.workspace_id), "project_status": project.status if project else "missing", "billing_status": billing_status, "companion_installed": companion_ready, "site_understanding_status": sync_status, "last_sync_at": latest_completed.completed_at.replace(microsecond=0).isoformat() if latest_completed and latest_completed.completed_at else None, "steps": steps, } async def handle_wordpress_chat_session_open( self, bearer_token: str, body: WordPressChatSessionOpenRequest, ) -> WordPressChatSessionOpenResponse: installation = await self._resolve_wordpress_installation_for_chat( bearer_token, body.installation_id, ) await self._require_wordpress_site_ready(installation) existing = await self._chat_sessions.get_active_for_installation(installation.id) if existing: return WordPressChatSessionOpenResponse( ok=True, session_id=str(existing.id), expires_at=existing.expires_at.replace(microsecond=0).isoformat(), poll_interval_ms=2000, ) expires_at = datetime.now(timezone.utc) + timedelta(hours=_CHAT_SESSION_TTL_HOURS) session = ExternalAdminChatSession( installation_id=installation.id, project_id=installation.project_id, site_url=body.site_url, status="active", expires_at=expires_at, ) await self._chat_sessions.create(session) return WordPressChatSessionOpenResponse( ok=True, session_id=str(session.id), expires_at=expires_at.replace(microsecond=0).isoformat(), poll_interval_ms=2000, ) async def handle_wordpress_chat_message_send( self, bearer_token: str, body: WordPressChatMessageSendRequest, ) -> WordPressChatMessageSendResponse: installation = await self._resolve_wordpress_installation_for_chat( bearer_token, body.installation_id, ) await self._require_wordpress_site_ready(installation) idempotency_key = (body.idempotency_key or "").strip() or None # Idempotency: return existing job if we've seen this key before. if idempotency_key: existing_job = await self._chat_jobs.get_by_idempotency_key( installation.id, idempotency_key ) if existing_job: return await self._wordpress_chat_job_send_response(existing_job) try: session_id = UUID(body.session_id) except ValueError as exc: raise ValidationError("invalid_session_id") from exc session = await self._require_active_wordpress_chat_session(session_id, installation) message = body.message.strip() if not message: raise ValidationError("message_required") user_msg = ExternalAdminChatMessage( session_id=session.id, installation_id=installation.id, role="user", content=message, ) await self._chat_messages.create(user_msg) job = ExternalAdminChatJob( session_id=session.id, installation_id=installation.id, status=CHAT_JOB_STATUS_PENDING, user_message_id=user_msg.id, idempotency_key=idempotency_key, ) await self._chat_jobs.create(job) from src.workers.enqueue import enqueue_external_admin_chat_job await enqueue_external_admin_chat_job(str(job.id), defer_seconds=2.0) return await self._wordpress_chat_job_send_response(job) async def handle_wordpress_chat_message_list( self, bearer_token: str, body: WordPressChatMessageListRequest, ) -> WordPressChatMessageListResponse: installation = await self._resolve_wordpress_installation_for_chat(bearer_token, body.installation_id) try: session_id = UUID(body.session_id) except ValueError as exc: raise ValidationError("invalid_session_id") from exc session = await self._chat_sessions.get(session_id) if ( not session or session.installation_id != installation.id or session.project_id != installation.project_id ): raise ValidationError("invalid_chat_session") after_id: Optional[UUID] = None if body.cursor: try: after_id = UUID(body.cursor) except ValueError: pass if after_id: rows = await self._chat_messages.list_by_session(session_id, after_message_id=after_id, limit=body.limit) else: rows = await self._chat_messages.list_recent_by_session(session_id, limit=body.limit) items = [_format_chat_message(r) for r in rows] next_cursor = str(rows[-1].id) if len(rows) >= body.limit else None return WordPressChatMessageListResponse(ok=True, messages=items, next_cursor=next_cursor) async def handle_wordpress_chat_job_poll( self, bearer_token: str, body: WordPressChatJobPollRequest, ) -> WordPressChatJobPollResponse: installation = await self._resolve_wordpress_installation_for_chat(bearer_token, body.installation_id) try: job_id = UUID(body.job_id) except ValueError as exc: raise NotFoundError("chat_job_not_found") from exc job = await self._chat_jobs.get(job_id) if not job or job.installation_id != installation.id: raise NotFoundError("chat_job_not_found") assistant_msg = None if job.assistant_message_id: msg_row = await self._chat_messages.get(job.assistant_message_id) if msg_row: assistant_msg = _format_chat_message(msg_row) return WordPressChatJobPollResponse( ok=True, job_id=str(job.id), status=job.status, assistant_message=assistant_msg, action_session_id=str(job.action_session_id) if job.action_session_id else None, error=job.error, ) async def _wordpress_chat_job_send_response( self, job: ExternalAdminChatJob, ) -> WordPressChatMessageSendResponse: assistant_msg = None if job.assistant_message_id: msg_row = await self._chat_messages.get(job.assistant_message_id) if msg_row: assistant_msg = _format_chat_message(msg_row) return WordPressChatMessageSendResponse( ok=True, job_id=str(job.id), status=job.status, user_message_id=str(job.user_message_id) if job.user_message_id else None, assistant_message=assistant_msg, ) async def _require_active_wordpress_chat_session( self, session_id: UUID, installation: ExternalAdminInstallation, ) -> ExternalAdminChatSession: session = await self._chat_sessions.get(session_id) if ( not session or session.installation_id != installation.id or session.project_id != installation.project_id ): raise ValidationError("invalid_chat_session") if session.status != "active" or session.expires_at < datetime.now(timezone.utc): raise ValidationError("chat_session_expired") return session async def _require_wordpress_site_ready( self, installation: ExternalAdminInstallation, ) -> None: readiness = await self._wordpress_site_readiness(installation) if not readiness["ready"]: raise ValidationError(f"site_not_ready:{readiness['next_action']}") async def handle_wordpress_site_understanding_start( self, bearer_token: str, body: WordPressSiteUnderstandingSyncRequest, ) -> WordPressSiteUnderstandingSyncResponse: installation = await self._resolve_wordpress_installation_for_chat( bearer_token, body.installation_id, ) run = SiteUnderstandingRun( installation_id=installation.id, project_id=installation.project_id, platform="wordpress", status=SITE_UNDERSTANDING_STATUS_QUEUED, reason=body.reason or "manual", triggered_by="plugin", summary_json=self._wordpress_site_understanding_payload_summary(body.payload), payload_json=body.payload, ) await self._su_runs.create(run) from src.workers.enqueue import enqueue_wordpress_site_understanding_run await enqueue_wordpress_site_understanding_run(str(run.id), defer_seconds=2.0) return WordPressSiteUnderstandingSyncResponse( ok=True, sync_id=str(run.id), status=run.status, message="Site Understanding sync queued.", ) def _wordpress_site_understanding_payload_summary(self, payload: Optional[dict]) -> dict: if not payload: return { "message": "Site Understanding sync queued without inline payload.", "payload_received": False, } content = payload.get("content") if isinstance(payload.get("content"), dict) else {} inventory = payload.get("inventory") if isinstance(payload.get("inventory"), dict) else {} config = payload.get("config") if isinstance(payload.get("config"), dict) else {} db_changes = payload.get("db_changes") if isinstance(payload.get("db_changes"), dict) else {} entities = content.get("entities") if isinstance(content.get("entities"), list) else [] plugins = inventory.get("plugins") if isinstance(inventory.get("plugins"), list) else [] themes = inventory.get("themes") if isinstance(inventory.get("themes"), list) else [] changes = db_changes.get("unsynced") if isinstance(db_changes.get("unsynced"), list) else [] return { "message": "Site Understanding payload received; processing worker not yet wired.", "payload_received": True, "schema_version": payload.get("schema_version"), "generated_at": payload.get("generated_at"), "wp_config_available": bool(config.get("available")), "wp_config_sanitized_sha256": config.get("sanitized_sha256"), "entity_count": len(entities), "plugin_count": len(plugins), "theme_count": len(themes), "unsynced_change_count": len(changes), "content_mode": content.get("mode"), "db_change_mode": db_changes.get("mode"), } async def handle_wordpress_site_understanding_status( self, bearer_token: str, body: WordPressSiteUnderstandingStatusRequest, ) -> WordPressSiteUnderstandingStatusResponse: installation = await self._resolve_wordpress_installation_for_chat( bearer_token, body.installation_id, ) run = await self._su_runs.get_latest_for_installation(installation.id) latest_completed = await self._su_runs.get_latest_completed_for_installation( installation.id ) readiness = await self._wordpress_site_readiness(installation) if not run: return WordPressSiteUnderstandingStatusResponse( ok=True, status="not_started", sync_id=None, last_sync_at=None, message="Site Understanding has not run yet.", site_ready=readiness["ready"], readiness=readiness, ) return WordPressSiteUnderstandingStatusResponse( ok=True, status=run.status, sync_id=str(run.id), last_sync_at=latest_completed.completed_at.replace(microsecond=0).isoformat() if latest_completed and latest_completed.completed_at else None, message=run.summary_json.get("message") if run.summary_json else None, site_ready=readiness["ready"], readiness=readiness, ) async def _shopify_store_readiness( self, installation: ExternalAdminInstallation, ) -> dict: project = await self._projects.get_by_id(installation.project_id) latest_run = await self._su_runs.get_latest_for_installation(installation.id) latest_completed = await self._su_runs.get_latest_completed_for_installation( installation.id ) project_ready = bool(project and project.status == "ready") connection_ready = installation.status == INSTALLATION_STATUS_CONNECTED credential_ready = bool(await self._credentials.get_active(installation.id, "shopify_access_token")) sync_ready = bool(latest_completed and latest_completed.completed_at) billing_ready = True billing_message = "Billing access is active." billing_status = "unknown" if self._billing_policy: try: billing_status = await self._billing_policy.get_billing_status(installation.workspace_id) await self._billing_policy.require_active_billing(installation.workspace_id) except PaymentRequiredError as exc: billing_ready = False billing_status = getattr(exc, "billing_status", None) or billing_status billing_message = "Active billing or a valid free plan is required before assistant actions can run." sync_status = latest_run.status if latest_run else "not_started" sync_running = sync_status in {"queued", "running"} sync_failed = sync_status == "failed" steps = [ { "key": "connection", "label": "Shopify connection", "status": "complete" if connection_ready else "blocked", "required": True, "blocking": not connection_ready, "message": "This Shopify store is connected." if connection_ready else "Reconnect this Shopify store to P2HS.", }, { "key": "billing", "label": "P2HS billing", "status": "complete" if billing_ready else "blocked", "required": True, "blocking": not billing_ready, "message": billing_message, }, { "key": "project", "label": "P2HS project", "status": "complete" if project_ready else "pending", "required": True, "blocking": not project_ready, "message": "The P2HS project workspace is ready." if project_ready else "P2HS is still preparing the project workspace.", }, { "key": "shopify_access", "label": "Shopify API access", "status": "complete" if credential_ready else "blocked", "required": True, "blocking": not credential_ready, "message": "Shopify Admin API access is connected." if credential_ready else "Reconnect the Shopify app so P2HS can receive Admin API access.", }, { "key": "site_understanding", "label": "First store sync", "status": ( "complete" if sync_ready else "running" if sync_running else "failed" if sync_failed else "pending" ), "required": True, "blocking": not sync_ready, "message": "Store Understanding has completed." if sync_ready else "Store Understanding is running." if sync_running else "The latest Store Understanding sync failed. Run sync again." if sync_failed else "Run the first Store Understanding sync before chatting.", }, ] blocking_reasons = [step["key"] for step in steps if step["blocking"]] ready = not blocking_reasons next_action = "ready" if not connection_ready: next_action = "reconnect" elif not billing_ready: next_action = "resolve_billing" elif not project_ready: next_action = "wait_for_project" elif not credential_ready: next_action = "reconnect_shopify" elif sync_running: next_action = "wait_for_sync" elif not sync_ready: next_action = "sync_store" return { "ready": ready, "status": "ready" if ready else "setup_required", "next_action": next_action, "blocking_reasons": blocking_reasons, "project_id": str(installation.project_id), "workspace_id": str(installation.workspace_id), "project_status": project.status if project else "missing", "billing_status": billing_status, "shopify_access_connected": credential_ready, "site_understanding_status": sync_status, "last_sync_at": latest_completed.completed_at.replace(microsecond=0).isoformat() if latest_completed and latest_completed.completed_at else None, "steps": steps, } async def handle_shopify_heartbeat( self, bearer_token: str, body: ShopifyHeartbeatRequest, ) -> ShopifyHeartbeatResponse: installation = await self._resolve_shopify_installation_for_chat( bearer_token, body.installation_id, body.shop_domain, ) installation.last_heartbeat_at = datetime.now(timezone.utc) installation.shop_domain = self._normalize_shop_domain(body.shop_domain) await self._installations.save(installation) readiness = await self._shopify_store_readiness(installation) return ShopifyHeartbeatResponse( ok=True, status=installation.status, installation_id=str(installation.id), shop_domain=installation.shop_domain or self._normalize_shop_domain(body.shop_domain), site_ready=readiness["ready"], readiness=readiness, ) async def handle_shopify_chat_session_open( self, bearer_token: str, body: ShopifyChatSessionOpenRequest, ) -> ShopifyChatSessionOpenResponse: installation = await self._resolve_shopify_installation_for_chat( bearer_token, body.installation_id, body.shop_domain, ) await self._require_shopify_store_ready(installation) existing = await self._chat_sessions.get_active_for_installation(installation.id) if existing: return ShopifyChatSessionOpenResponse( ok=True, session_id=str(existing.id), expires_at=existing.expires_at.replace(microsecond=0).isoformat(), poll_interval_ms=2000, ) expires_at = datetime.now(timezone.utc) + timedelta(hours=_CHAT_SESSION_TTL_HOURS) session = ExternalAdminChatSession( installation_id=installation.id, project_id=installation.project_id, site_url=self._normalize_shop_domain(body.shop_domain), status="active", expires_at=expires_at, ) await self._chat_sessions.create(session) return ShopifyChatSessionOpenResponse( ok=True, session_id=str(session.id), expires_at=expires_at.replace(microsecond=0).isoformat(), poll_interval_ms=2000, ) async def handle_shopify_chat_message_send( self, bearer_token: str, body: ShopifyChatMessageSendRequest, ) -> ShopifyChatMessageSendResponse: installation = await self._resolve_shopify_installation_for_chat( bearer_token, body.installation_id, body.shop_domain, ) await self._require_shopify_store_ready(installation) idempotency_key = (body.idempotency_key or "").strip() or None if idempotency_key: existing_job = await self._chat_jobs.get_by_idempotency_key( installation.id, idempotency_key ) if existing_job: return await self._shopify_chat_job_send_response(existing_job) try: session_id = UUID(body.session_id) except ValueError as exc: raise ValidationError("invalid_session_id") from exc session = await self._require_active_shopify_chat_session(session_id, installation) message = body.message.strip() if not message: raise ValidationError("message_required") user_msg = ExternalAdminChatMessage( session_id=session.id, installation_id=installation.id, role="user", content=message, ) await self._chat_messages.create(user_msg) job = ExternalAdminChatJob( session_id=session.id, installation_id=installation.id, status=CHAT_JOB_STATUS_PENDING, user_message_id=user_msg.id, idempotency_key=idempotency_key, ) await self._chat_jobs.create(job) from src.workers.enqueue import enqueue_external_admin_chat_job await enqueue_external_admin_chat_job(str(job.id), defer_seconds=2.0) return await self._shopify_chat_job_send_response(job) async def handle_shopify_chat_message_list( self, bearer_token: str, body: ShopifyChatMessageListRequest, ) -> ShopifyChatMessageListResponse: installation = await self._resolve_shopify_installation_for_chat( bearer_token, body.installation_id, body.shop_domain ) try: session_id = UUID(body.session_id) except ValueError as exc: raise ValidationError("invalid_session_id") from exc session = await self._chat_sessions.get(session_id) if ( not session or session.installation_id != installation.id or session.project_id != installation.project_id ): raise ValidationError("invalid_chat_session") after_id: Optional[UUID] = None if body.cursor: try: after_id = UUID(body.cursor) except ValueError: pass if after_id: rows = await self._chat_messages.list_by_session(session_id, after_message_id=after_id, limit=body.limit) else: rows = await self._chat_messages.list_recent_by_session(session_id, limit=body.limit) items = [_format_shopify_chat_message(r) for r in rows] next_cursor = str(rows[-1].id) if len(rows) >= body.limit else None return ShopifyChatMessageListResponse(ok=True, messages=items, next_cursor=next_cursor) async def handle_shopify_chat_job_poll( self, bearer_token: str, body: ShopifyChatJobPollRequest, ) -> ShopifyChatJobPollResponse: installation = await self._resolve_shopify_installation_for_chat( bearer_token, body.installation_id, body.shop_domain ) try: job_id = UUID(body.job_id) except ValueError as exc: raise NotFoundError("chat_job_not_found") from exc job = await self._chat_jobs.get(job_id) if not job or job.installation_id != installation.id: raise NotFoundError("chat_job_not_found") assistant_msg = None if job.assistant_message_id: msg_row = await self._chat_messages.get(job.assistant_message_id) if msg_row: assistant_msg = _format_shopify_chat_message(msg_row) return ShopifyChatJobPollResponse( ok=True, job_id=str(job.id), status=job.status, assistant_message=assistant_msg, action_session_id=str(job.action_session_id) if job.action_session_id else None, error=job.error, ) async def _shopify_chat_job_send_response( self, job: ExternalAdminChatJob, ) -> ShopifyChatMessageSendResponse: assistant_msg = None if job.assistant_message_id: msg_row = await self._chat_messages.get(job.assistant_message_id) if msg_row: assistant_msg = _format_shopify_chat_message(msg_row) return ShopifyChatMessageSendResponse( ok=True, job_id=str(job.id), status=job.status, user_message_id=str(job.user_message_id) if job.user_message_id else None, assistant_message=assistant_msg, ) async def _require_active_shopify_chat_session( self, session_id: UUID, installation: ExternalAdminInstallation, ) -> ExternalAdminChatSession: session = await self._chat_sessions.get(session_id) if ( not session or session.installation_id != installation.id or session.project_id != installation.project_id ): raise ValidationError("invalid_chat_session") if session.status != "active" or session.expires_at < datetime.now(timezone.utc): raise ValidationError("chat_session_expired") return session async def _require_shopify_store_ready( self, installation: ExternalAdminInstallation, ) -> None: readiness = await self._shopify_store_readiness(installation) if not readiness["ready"]: raise ValidationError(f"store_not_ready:{readiness['next_action']}") async def handle_shopify_site_understanding_start( self, bearer_token: str, body: ShopifySiteUnderstandingSyncRequest, ) -> ShopifySiteUnderstandingSyncResponse: installation = await self._resolve_shopify_installation_for_chat( bearer_token, body.installation_id, body.shop_domain, ) run = SiteUnderstandingRun( installation_id=installation.id, project_id=installation.project_id, platform="shopify", status=SITE_UNDERSTANDING_STATUS_QUEUED, reason=body.reason or "manual", triggered_by="shopify_app", summary_json=self._shopify_site_understanding_payload_summary(body.payload), payload_json=body.payload, ) await self._su_runs.create(run) from src.workers.enqueue import enqueue_shopify_site_understanding_run await enqueue_shopify_site_understanding_run(str(run.id), defer_seconds=2.0) return ShopifySiteUnderstandingSyncResponse( ok=True, sync_id=str(run.id), status=run.status, message="Store Understanding sync queued.", ) async def handle_shopify_site_understanding_status( self, bearer_token: str, body: ShopifySiteUnderstandingStatusRequest, ) -> ShopifySiteUnderstandingStatusResponse: installation = await self._resolve_shopify_installation_for_chat( bearer_token, body.installation_id, body.shop_domain, ) run = await self._su_runs.get_latest_for_installation(installation.id) latest_completed = await self._su_runs.get_latest_completed_for_installation( installation.id ) readiness = await self._shopify_store_readiness(installation) if not run: return ShopifySiteUnderstandingStatusResponse( ok=True, status="not_started", sync_id=None, last_sync_at=None, message="Store Understanding has not run yet.", site_ready=readiness["ready"], readiness=readiness, ) return ShopifySiteUnderstandingStatusResponse( ok=True, status=run.status, sync_id=str(run.id), last_sync_at=latest_completed.completed_at.replace(microsecond=0).isoformat() if latest_completed and latest_completed.completed_at else None, message=run.summary_json.get("message") if run.summary_json else None, site_ready=readiness["ready"], readiness=readiness, ) def _shopify_site_understanding_payload_summary(self, payload: Optional[dict]) -> dict: if not payload: return { "message": "Store Understanding sync queued without inline payload.", "payload_received": False, } store = payload.get("store") if isinstance(payload.get("store"), dict) else {} products = payload.get("products") if isinstance(payload.get("products"), list) else [] collections = payload.get("collections") if isinstance(payload.get("collections"), list) else [] pages = payload.get("pages") if isinstance(payload.get("pages"), list) else [] theme = payload.get("theme") if isinstance(payload.get("theme"), dict) else {} changes = payload.get("changes") if isinstance(payload.get("changes"), list) else [] return { "message": "Store Understanding payload received.", "payload_received": True, "schema_version": payload.get("schema_version"), "generated_at": payload.get("generated_at"), "shop_domain": store.get("domain") or store.get("myshopify_domain"), "product_count": len(products), "collection_count": len(collections), "page_count": len(pages), "theme_asset_count": len(theme.get("assets") or {}) if isinstance(theme.get("assets"), dict) else 0, "unsynced_change_count": len(changes), } async def handle_wordpress_companion_install( self, bearer_token: str, body: CompanionInstallRequest, ) -> CompanionInstallResponse: if not get_settings().wp_companion_install_enabled: raise ForbiddenError("companion_install_disabled") installation = await self._resolve_wordpress_installation_for_chat( bearer_token, body.installation_id, ) update = await self._check_companion_update(installation) if not update: raise ValidationError("companion_install_package_not_available") return CompanionInstallResponse( ok=True, download_url=update["url"], signature=update["signature"], version=str(update.get("version") or body.requested_version or ""), sha256=str(update.get("sha256") or ""), ) async def build_wordpress_critical_flow_url( self, bearer_token: str, body: WordPressHostedUrlRequest, ) -> WordPressHostedUrlResponse: installation = await self._resolve_wordpress_installation_for_hosted_url(bearer_token, body) app_url = get_settings().app_url.rstrip("/") query_args = { "external_admin": "critical_flow", "flow": body.flow or "", "installation_id": str(installation.id), "site_url": body.site_url, } if body.return_url: query_args["return_url"] = body.return_url query = urlencode(query_args) return WordPressHostedUrlResponse( ok=True, url=f"{app_url}/projects/{installation.project_id}?{query}", ) # ── WordPress heartbeat ─────────────────────────────────────────────────── async def handle_wordpress_heartbeat( self, bearer_token: str, body: WordPressHeartbeatRequest, ) -> WordPressHeartbeatResponse: installation = await self._resolve_installation_by_token(bearer_token) if str(installation.id) != body.installation_id: raise ForbiddenError("installation_id_mismatch") if installation.status == INSTALLATION_STATUS_DISCONNECTED: raise ForbiddenError("installation_disconnected") now = datetime.now(timezone.utc) installation.last_heartbeat_at = now installation.site_url = body.site_url # Tamper detection runs BEFORE updating stored plugin hash so the # stored value acts as a stable baseline. Companion hash is checked # against the live archive (not the stored value), so update order # for companion doesn't matter. tamper = self._check_tamper( installation, body.plugin_version, body.plugin_file_hash, body.companion_file_hash ) if body.companion_version: installation.companion_version = body.companion_version if body.companion_file_hash: installation.companion_file_hash = body.companion_file_hash # Plugin hash baseline: update only on version change or first heartbeat. # Keeping the baseline stable is what makes same-version hash changes detectable. if body.plugin_version != installation.plugin_version: installation.plugin_version = body.plugin_version if body.plugin_file_hash: installation.plugin_file_hash = body.plugin_file_hash elif not installation.plugin_file_hash and body.plugin_file_hash: installation.plugin_version = body.plugin_version installation.plugin_file_hash = body.plugin_file_hash if tamper and installation.status != INSTALLATION_STATUS_TAMPERED: installation.status = INSTALLATION_STATUS_TAMPERED installation.tampered_at = now elif not tamper and installation.status == INSTALLATION_STATUS_TAMPERED: pass # Remain tampered until an admin reviews + resets. elif not tamper and installation.status != INSTALLATION_STATUS_CONNECTED: installation.status = INSTALLATION_STATUS_CONNECTED if not installation.connected_at: installation.connected_at = now await self._installations.save(installation) companion_update = await self._check_companion_update(installation) readiness = await self._wordpress_site_readiness(installation) await self._maybe_provision_wordpress_project(installation, readiness) return WordPressHeartbeatResponse( ok=True, status=installation.status, installation_id=str(installation.id), companion_update_available=companion_update is not None, companion_update_url=companion_update["url"] if companion_update else None, companion_update_signature=companion_update["signature"] if companion_update else None, companion_update_version=companion_update["version"] if companion_update else None, companion_update_sha256=companion_update["sha256"] if companion_update else None, site_ready=readiness["ready"], readiness=readiness, ) async def _maybe_provision_wordpress_project( self, installation: ExternalAdminInstallation, readiness: dict, ) -> None: """Backstop: ensure a connected WordPress project gets GitHub provisioning. Heals projects whose initial setup job failed, or that predate the WordPress setup pipeline — including installs older than the recovery scanner's age window, which it deliberately skips. Heartbeats are naturally throttled (admin-load at most once / 6h plus the daily cron) and ``setup_project_repo`` is idempotent (it short-circuits once setup_stage is ``completed``), so re-enqueuing here is safe. A short age guard avoids racing a brand-new project's initial setup job, which the transactional outbox already enqueued. """ if readiness.get("project_status") == "ready": return from src.domain.projects.models import AI_ASSIST_WORDPRESS_PROJECT_TYPE project = await self._projects.get_by_id(installation.project_id) if ( not project or project.deleted_at is not None or project.type != AI_ASSIST_WORDPRESS_PROJECT_TYPE or project.status == "ready" ): return created_at = project.created_at if created_at is not None: if created_at.tzinfo is None: created_at = created_at.replace(tzinfo=timezone.utc) if datetime.now(timezone.utc) - created_at < timedelta(minutes=10): return from src.core.database import async_session_factory from src.domain.projects.repository import ProjectRepository from src.workers.enqueue import enqueue_project_setup # Surface a "recovering" state in the dashboard. Commit it in its own # transaction *before* enqueueing so a fast worker that finishes with # status="ready" is not clobbered back to "creating" by the request's # end-of-cycle commit (mirrors the project_recovery worker's ordering). if project.status != "creating": async with async_session_factory() as db: await ProjectRepository(db).update(project.id, status="creating") await db.commit() await enqueue_project_setup(str(project.id)) # ── Shopify OAuth callback ──────────────────────────────────────────────── async def handle_shopify_oauth_callback( self, body: ShopifyOAuthCallbackRequest, ) -> dict: installation = await self._resolve_shopify_installation_from_state(body.state, body.shop) if not installation: raise NotFoundError("external_admin_installation_not_found") access_token = await self._exchange_shopify_oauth_code(body.shop, body.code) now = datetime.now(timezone.utc) installation.shop_domain = self._normalize_shop_domain(body.shop) installation.status = INSTALLATION_STATUS_CONNECTED installation.connected_at = now installation.disconnected_at = None # The per-installation credential the BFF uses from now on, minted here (as WordPress does # at connect) and returned once. Every (re)install replaces it, which retires the previous # one. Only its hash and an encrypted copy are kept. raw_token = secrets.token_urlsafe(32) installation.installation_token_enc = self._encrypt(raw_token) installation.metadata_json = { **(installation.metadata_json or {}), "installation_token_hash": self._token_hash(raw_token), } await self._installations.save(installation) existing = await self._credentials.get_active(installation.id, "shopify_access_token") if existing: existing.status = CREDENTIAL_STATUS_ROTATED existing.rotated_at = now await self._credentials.save(existing) max_version = await self._credentials.get_max_version(installation.id, "shopify_access_token") await self._credentials.create( ExternalCredential( installation_id=installation.id, project_id=installation.project_id, credential_type="shopify_access_token", version=max_version + 1, value_enc=self._encrypt(access_token), status=CREDENTIAL_STATUS_ACTIVE, ) ) await self._register_shopify_webhooks(installation.shop_domain, access_token) return { "ok": True, "installation_id": str(installation.id), "shop_domain": installation.shop_domain, "redirect_url": f"https://{installation.shop_domain}/admin/apps/p2hs-ai", "installation_token": raw_token, } async def handle_shopify_webhook(self, body: ShopifyWebhookEnvelope) -> dict: if body.topic in SHOPIFY_PRIVACY_TOPICS: return await self._handle_shopify_privacy_webhook(body) normalized_shop = self._normalize_shop_domain(body.shop_domain) installation = await self._installations.get_by_shop_domain(normalized_shop) if not installation: raise NotFoundError("external_admin_installation_not_found") metadata = dict(installation.metadata_json or {}) metadata.setdefault("shopify_webhooks", []) metadata["shopify_webhooks"] = [ *metadata["shopify_webhooks"][-19:], { "topic": body.topic, "received_at": body.received_at, }, ] installation.metadata_json = metadata if body.topic == "app/uninstalled": installation.status = INSTALLATION_STATUS_DISCONNECTED installation.disconnected_at = datetime.now(timezone.utc) await self._installations.save(installation) return {"ok": True, "status": installation.status} async def _handle_shopify_privacy_webhook(self, body: ShopifyWebhookEnvelope) -> dict: """Shopify mandatory privacy webhooks (GDPR). * ``shop/redact``: erase all data held for the shop (Shopify sends it ~48h after uninstall) and keep only a tombstone. * ``customers/redact`` / ``customers/data_request``: Core stores nothing keyed by a Shopify customer (Site Understanding syncs products, collections, pages and theme; chat is merchant-authored), so there is nothing to erase or export. The request is recorded with that outcome. The payload is deliberately NOT stored. Every topic answers 200 and is idempotent: Shopify retries, and an unknown or already-redacted shop means there is nothing left to do (never a 404, which would make Shopify retry forever). """ normalized_shop = self._normalize_shop_domain(body.shop_domain) installation = await self._installations.get_by_shop_domain(normalized_shop) if not installation: return {"ok": True, "topic": body.topic, "status": "no_data_held"} now = datetime.now(timezone.utc) record = {"topic": body.topic, "received_at": body.received_at, "handled_at": now.isoformat()} if body.topic == "shop/redact": removed = await self._installations.redact_installation_data(installation.id) installation.installation_token_enc = None installation.shop_domain = None installation.site_url = None installation.last_error = None installation.companion_version = None installation.plugin_version = None installation.status = INSTALLATION_STATUS_REDACTED installation.disconnected_at = installation.disconnected_at or now # Tombstone: audit of the erasure only. Shopify webhook history is dropped too. installation.metadata_json = { "privacy_requests": [{**record, "outcome": "shop_data_erased", "removed": removed}] } await self._installations.save(installation) logger.info( "shopify_shop_redacted installation_id=%s removed=%s", installation.id, removed ) return {"ok": True, "topic": body.topic, "status": "shop_data_erased"} # customers/redact, customers/data_request metadata = dict(installation.metadata_json or {}) metadata["privacy_requests"] = [ *list(metadata.get("privacy_requests") or [])[-49:], {**record, "outcome": "no_customer_data_held"}, ] installation.metadata_json = metadata await self._installations.save(installation) return {"ok": True, "topic": body.topic, "status": "no_customer_data_held"} async def resolve_shopify_connect_target( self, workspace_id: UUID, shop_domain: str, ) -> Optional[UUID]: """Which existing project a Shopify connect for ``shop_domain`` should reuse. * this workspace already has an installation for the shop (connected, or uninstalled and now being reinstalled) -> reuse its project, so a reinstall does not create a second project and installation for the same shop; * another workspace has it connected -> ``ConflictError`` (before any project is created); * otherwise -> None, a new project is created. """ normalized_shop = self._normalize_shop_domain(shop_domain) installations = await self._installations.list_by_shop_domain(normalized_shop) for installation in installations: if installation.workspace_id == workspace_id: return installation.project_id if any(i.status == INSTALLATION_STATUS_CONNECTED for i in installations): raise ConflictError("shop_already_connected") return None async def create_shopify_pending_installation( self, project_id: UUID, workspace_id: UUID, shop_domain: str, ) -> ShopifyConnectProjectResponse: project = await self._projects.get_by_id(project_id) if not project or project.workspace_id != workspace_id: raise ForbiddenError("project_not_in_workspace") if project.type != "ai_assist_shopify": raise ForbiddenError("project_type_mismatch") normalized_shop = self._normalize_shop_domain(shop_domain) # A shop is connected to one project at a time. Another project's live installation for # this shop blocks this one; a dead one (uninstalled, errored) does not. for other in await self._installations.list_by_shop_domain(normalized_shop): if other.project_id != project_id and other.status == INSTALLATION_STATUS_CONNECTED: raise ConflictError("shop_already_connected") installation = await self._installations.get_by_project(project_id) if not installation: installation = ExternalAdminInstallation( project_id=project_id, workspace_id=workspace_id, installation_type=INSTALLATION_TYPE_SHOPIFY, status=INSTALLATION_STATUS_PENDING, shop_domain=normalized_shop, ) await self._installations.create(installation) elif installation.installation_type != INSTALLATION_TYPE_SHOPIFY: raise ForbiddenError("installation_type_mismatch") else: installation.shop_domain = normalized_shop if installation.status != INSTALLATION_STATUS_CONNECTED: # Reinstall after an uninstall: the same installation starts over rather than a # second one appearing for the same shop. installation.status = INSTALLATION_STATUS_PENDING installation.disconnected_at = None await self._installations.save(installation) return ShopifyConnectProjectResponse( ok=True, project_id=str(project_id), installation_id=str(installation.id), shop_domain=normalized_shop, ) # ── Settings ───────────────────────────────────────────────────────────── async def get_settings(self, project_id: UUID, workspace_id: UUID) -> AssistantProjectSettings: await self._assert_project_in_workspace(project_id, workspace_id) settings = await self._settings.get(project_id) if not settings: settings = AssistantProjectSettings(project_id=project_id) await self._settings.upsert(settings) return settings async def update_settings( self, project_id: UUID, workspace_id: UUID, update: AssistantProjectSettingsUpdate, ) -> AssistantProjectSettings: settings = await self.get_settings(project_id, workspace_id) for field, value in update.model_dump(exclude_none=True).items(): setattr(settings, field, value) return await self._settings.upsert(settings) # ── Credentials ─────────────────────────────────────────────────────────── async def list_credentials(self, project_id: UUID, workspace_id: UUID) -> list[ExternalCredential]: installation = await self.get_installation(project_id, workspace_id) return await self._credentials.list_by_installation(installation.id) async def rotate_credential( self, project_id: UUID, workspace_id: UUID, credential_type: str, rotated_by_user_id: UUID, ) -> ExternalCredential: installation = await self.get_installation(project_id, workspace_id) existing = await self._credentials.get_active(installation.id, credential_type) if existing: existing.status = CREDENTIAL_STATUS_ROTATED existing.rotated_at = datetime.now(timezone.utc) existing.rotated_by_user_id = rotated_by_user_id await self._credentials.save(existing) max_version = await self._credentials.get_max_version(installation.id, credential_type) new_raw = secrets.token_urlsafe(32) new_cred = ExternalCredential( installation_id=installation.id, project_id=project_id, credential_type=credential_type, version=max_version + 1, value_enc=self._encrypt(new_raw), status=CREDENTIAL_STATUS_ACTIVE, ) return await self._credentials.create(new_cred) # ── Action sessions ──────────────────────────────────────────────────────── async def list_action_sessions( self, project_id: UUID, workspace_id: UUID, status: Optional[str] = None, limit: int = 50, offset: int = 0, ) -> tuple[list[AssistantActionSession], int]: await self._assert_project_in_workspace(project_id, workspace_id) return await self._action_sessions.list_by_project(project_id, status=status, limit=limit, offset=offset) async def get_action_session( self, project_id: UUID, workspace_id: UUID, session_id: UUID, ) -> AssistantActionSession: await self._assert_project_in_workspace(project_id, workspace_id) row = await self._action_sessions.get(session_id) if not row or row.project_id != project_id: raise NotFoundError("action_session_not_found") return row async def approve_action_session( self, project_id: UUID, workspace_id: UUID, session_id: UUID, reviewer_user_id: UUID, note: Optional[str] = None, ) -> AssistantActionSession: session = await self.get_action_session(project_id, workspace_id, session_id) if session.status != "pending": raise ValidationError(f"action_session_not_pending: current status is {session.status!r}") now = datetime.now(timezone.utc) if session.expires_at < now: session.status = "expired" await self._action_sessions.save(session) raise ValidationError("action_session_expired") session.status = ACTION_SESSION_STATUS_APPROVED session.reviewed_by_user_id = reviewer_user_id session.reviewed_at = now if note: session.result = {"reviewer_note": note} saved = await self._action_sessions.save(session) # Enqueue execution based on platform. plan = session.payload or {} if plan.get("operations") and session.installation_id: installation = await self._installations.get_by_id(session.installation_id) if installation and installation.installation_type == INSTALLATION_TYPE_SHOPIFY: from src.workers.enqueue import enqueue_execute_shopify_mutation await enqueue_execute_shopify_mutation(str(session.id)) else: from src.workers.enqueue import enqueue_execute_wp_mutation await enqueue_execute_wp_mutation(str(session.id)) return saved async def reject_action_session( self, project_id: UUID, workspace_id: UUID, session_id: UUID, reviewer_user_id: UUID, note: Optional[str] = None, ) -> AssistantActionSession: session = await self.get_action_session(project_id, workspace_id, session_id) if session.status != "pending": raise ValidationError(f"action_session_not_pending: current status is {session.status!r}") now = datetime.now(timezone.utc) session.status = ACTION_SESSION_STATUS_REJECTED session.reviewed_by_user_id = reviewer_user_id session.reviewed_at = now if note: session.result = {"reviewer_note": note} return await self._action_sessions.save(session) # ── Consents ────────────────────────────────────────────────────────────── async def list_consents(self, project_id: UUID, workspace_id: UUID) -> list[AssistantDataConsent]: await self._assert_project_in_workspace(project_id, workspace_id) return await self._consents.list_by_project(project_id) async def grant_consent( self, project_id: UUID, workspace_id: UUID, consent_type: str, granted_by_user_id: UUID, notes: Optional[str] = None, ) -> AssistantDataConsent: await self._assert_project_in_workspace(project_id, workspace_id) existing = await self._consents.get(project_id, consent_type) now = datetime.now(timezone.utc) if existing: existing.status = "granted" existing.granted_at = now existing.revoked_at = None existing.granted_by_user_id = granted_by_user_id existing.notes = notes return await self._consents.upsert(existing) consent = AssistantDataConsent( project_id=project_id, consent_type=consent_type, status="granted", granted_at=now, granted_by_user_id=granted_by_user_id, notes=notes, ) return await self._consents.upsert(consent) async def revoke_consent( self, project_id: UUID, workspace_id: UUID, consent_type: str, revoked_by_user_id: UUID, ) -> AssistantDataConsent: await self._assert_project_in_workspace(project_id, workspace_id) existing = await self._consents.get(project_id, consent_type) if not existing: raise NotFoundError("consent_not_found") now = datetime.now(timezone.utc) existing.status = "revoked" existing.revoked_at = now return await self._consents.upsert(existing) # ── Helpers ─────────────────────────────────────────────────────────────── async def _assert_project_in_workspace(self, project_id: UUID, workspace_id: UUID) -> None: project = await self._projects.get_by_id(project_id) if not project or project.workspace_id != workspace_id: raise ForbiddenError("project_not_in_workspace") async def _resolve_installation_by_token(self, bearer_token: str) -> ExternalAdminInstallation: installation = await self._installations.get_by_installation_token_hash(self._token_hash(bearer_token)) if not installation: # Backward-compatible lookup for early local installs that used the placeholder scheme. installation = await self._installations.get_by_installation_token(f"enc:{bearer_token}") if not installation: raise ForbiddenError("invalid_installation_token") return installation async def _resolve_wordpress_installation_for_hosted_url( self, bearer_token: str, body: WordPressHostedUrlRequest, ) -> ExternalAdminInstallation: installation = await self._resolve_installation_by_token(bearer_token) if str(installation.id) != body.installation_id: raise ForbiddenError("installation_id_mismatch") if installation.installation_type != INSTALLATION_TYPE_WORDPRESS: raise ForbiddenError("installation_type_mismatch") if installation.status == INSTALLATION_STATUS_DISCONNECTED: raise ForbiddenError("installation_disconnected") return installation async def _resolve_wordpress_installation_for_chat( self, bearer_token: str, installation_id: str, ) -> ExternalAdminInstallation: installation = await self._resolve_installation_by_token(bearer_token) if str(installation.id) != installation_id: raise ForbiddenError("installation_id_mismatch") if installation.installation_type != INSTALLATION_TYPE_WORDPRESS: raise ForbiddenError("installation_type_mismatch") if installation.status == INSTALLATION_STATUS_DISCONNECTED: raise ForbiddenError("installation_disconnected") return installation async def _resolve_shopify_installation_for_chat( self, bearer_token: str, installation_id: str, shop_domain: Optional[str] = None, ) -> ExternalAdminInstallation: installation = await self._resolve_installation_by_token(bearer_token) if str(installation.id) != installation_id: raise ForbiddenError("installation_id_mismatch") if installation.installation_type != INSTALLATION_TYPE_SHOPIFY: raise ForbiddenError("installation_type_mismatch") if shop_domain: normalized = self._normalize_shop_domain(shop_domain) if installation.shop_domain and installation.shop_domain != normalized: raise ForbiddenError("shop_domain_mismatch") if installation.status == INSTALLATION_STATUS_DISCONNECTED: raise ForbiddenError("installation_disconnected") return installation # ── WordPress apply pull model ──────────────────────────────────────────── async def handle_wordpress_apply_poll( self, bearer_token: str, body: WordPressCompanionMutationPollRequest, ) -> WordPressCompanionMutationPollResponse: return await self.handle_companion_mutation_poll(bearer_token, body) async def handle_wordpress_apply_report( self, bearer_token: str, body: WordPressCompanionMutationReportRequest, ) -> WordPressCompanionMutationReportResponse: return await self.handle_companion_mutation_report(bearer_token, body) async def handle_shopify_apply_poll( self, bearer_token: str, body: ShopifyApplyPollRequest, ) -> ShopifyApplyPollResponse: # Shopify mutations are executed directly by Core via the Admin API. # The BFF/app never executes mutations; this route is not part of the # Shopify apply model and always returns has_mutation=False. await self._resolve_shopify_installation_for_chat(bearer_token, body.installation_id) return ShopifyApplyPollResponse(ok=True, has_mutation=False) async def handle_shopify_apply_report( self, bearer_token: str, body: ShopifyApplyReportRequest, ) -> ShopifyApplyReportResponse: # Shopify mutations are applied by Core directly; the app never reports # apply results. This route is a no-op stub kept for API surface parity. await self._resolve_shopify_installation_for_chat(bearer_token, body.installation_id) return ShopifyApplyReportResponse( ok=True, session_id="", session_status="not_applicable", has_more=False, ) # Backward-compatible names for older gateway/plugin builds. async def handle_companion_mutation_poll( self, bearer_token: str, body: WordPressCompanionMutationPollRequest, ) -> WordPressCompanionMutationPollResponse: installation = await self._resolve_wordpress_installation_for_chat( bearer_token, body.installation_id ) await self._mutation_actions.reset_stale_running(installation.id) next_op = await self._mutation_actions.get_next_pending(installation.id) if not next_op: return WordPressCompanionMutationPollResponse(ok=True, has_mutation=False) next_op.status = WP_MUTATION_ACTION_STATUS_RUNNING await self._mutation_actions.save(next_op) return WordPressCompanionMutationPollResponse( ok=True, has_mutation=True, action_id=str(next_op.id), session_id=str(next_op.action_session_id), sequence=next_op.sequence_number, action=next_op.action, description=next_op.description, payload=next_op.payload or {}, correlation_id=str(next_op.id), ) async def handle_companion_mutation_report( self, bearer_token: str, body: WordPressCompanionMutationReportRequest, ) -> WordPressCompanionMutationReportResponse: installation = await self._resolve_wordpress_installation_for_chat( bearer_token, body.installation_id ) try: action_uuid = UUID(body.action_id) except ValueError: raise ForbiddenError("invalid_action_id") op = await self._mutation_actions.get_by_id(action_uuid) if not op or op.installation_id != installation.id: raise ForbiddenError("action_not_found") now = datetime.now(timezone.utc) op.executed_at = now op.result_json = body.result op.after_hash = _hash_result(body.result) if body.ok: op.status = WP_MUTATION_ACTION_STATUS_COMPLETED else: op.status = WP_MUTATION_FAILED op.error = str(body.result.get("error") or "companion_action_failed")[:500] await self._mutation_actions.save(op) await self._mutation_actions.mark_pending_as_skipped(op.action_session_id) session = await self._action_sessions.get(op.action_session_id) session_status = ACTION_SESSION_STATUS_EXECUTING has_more = False if not body.ok: session_status = ACTION_SESSION_STATUS_FAILED if session: session.status = session_status await self._action_sessions.save(session) else: await self._mutation_actions.save(op) remaining = await self._mutation_actions.count_pending_or_running_for_session( op.action_session_id ) has_more = remaining > 0 if not has_more and session: session_status = ACTION_SESSION_STATUS_APPLIED session.status = session_status session.applied_at = now await self._action_sessions.save(session) return WordPressCompanionMutationReportResponse( ok=True, session_id=str(op.action_session_id), session_status=session_status, has_more=has_more, ) async def _register_shopify_webhooks(self, shop_domain: str, access_token: str) -> None: from src.integrations.shopify_admin_client import ShopifyAdminClient bff_url = get_settings().shopify_app_bff_url if not bff_url: return topics = [ "app/uninstalled", "customers/data_request", "customers/redact", "shop/redact", "products/create", "products/update", "products/delete", "collections/create", "collections/update", "collections/delete", "pages/create", "pages/update", "pages/delete", "themes/create", "themes/update", "themes/delete", "shop/update", ] try: client = ShopifyAdminClient(shop_domain, access_token) results = await client.register_webhooks(bff_url, topics) import structlog structlog.get_logger().info( "shopify_webhooks_registered", shop=shop_domain, registered=[r["topic"] for r in results if r.get("action") in ("created", "exists")], failed=[r["topic"] for r in results if r.get("action") == "failed"], ) except Exception as exc: import structlog structlog.get_logger().warning( "shopify_webhook_registration_error", shop=shop_domain, error=str(exc), ) def _check_tamper( self, installation: ExternalAdminInstallation, plugin_version: Optional[str], plugin_hash: Optional[str], companion_hash: Optional[str], ) -> bool: # Companion: compare reported dir hash against expected hash from the live archive. if companion_hash: expected = _compute_companion_expected_hash() if expected and companion_hash != expected: return True # Plugin: same version + different hash from stored baseline = tamper. if ( plugin_hash and installation.plugin_file_hash and plugin_version and installation.plugin_version and plugin_version == installation.plugin_version and plugin_hash != installation.plugin_file_hash ): return True return False async def _check_companion_update(self, installation: ExternalAdminInstallation) -> Optional[dict]: package = resolve_wordpress_companion_package() if not package: return None installed_version = (installation.companion_version or "").strip() if installed_version: return None api_url = resolve_wp_plugin_api_url().rstrip("/") return { "url": f"{api_url}/v1/wp-ai/companion/packages/{package.version}/download", "signature": "", "version": package.version, "sha256": package.sha256, } def _encrypt(self, value: str) -> str: return encrypt_value(value) def _token_hash(self, value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def _normalize_shop_domain(self, value: str) -> str: shop = value.strip().lower().removeprefix("https://").removeprefix("http://").split("/")[0] if "." not in shop: shop = f"{shop}.myshopify.com" if not shop.endswith(".myshopify.com"): raise ValidationError("invalid_shop_domain") return shop async def _resolve_shopify_installation_from_state( self, state: str, shop_domain: str, ) -> ExternalAdminInstallation: state = state.strip() normalized_shop = self._normalize_shop_domain(shop_domain) try: installation_uuid = UUID(state) except ValueError: installation_uuid = None installation = await self._installations.get_by_id(installation_uuid) if installation_uuid else None if not installation: installation = await self._installations.get_by_shop_domain(normalized_shop) if not installation: raise NotFoundError("external_admin_installation_not_found") if installation.installation_type != INSTALLATION_TYPE_SHOPIFY: raise ForbiddenError("installation_type_mismatch") if installation.shop_domain and self._normalize_shop_domain(installation.shop_domain) != normalized_shop: raise ForbiddenError("shop_domain_mismatch") return installation async def _exchange_shopify_oauth_code(self, shop_domain: str, code: str) -> str: settings = get_settings() if not settings.shopify_platform_client_id or not settings.shopify_platform_client_secret: raise ValidationError("shopify_oauth_not_configured") shop = self._normalize_shop_domain(shop_domain) async with httpx.AsyncClient(timeout=20.0) as client: response = await client.post( f"https://{shop}/admin/oauth/access_token", json={ "client_id": settings.shopify_platform_client_id, "client_secret": settings.shopify_platform_client_secret, "code": code, }, ) if response.status_code >= 400: raise ForbiddenError("shopify_oauth_exchange_failed") data = response.json() token = str(data.get("access_token") or "") if not token: raise ForbiddenError("shopify_oauth_missing_access_token") return token