"""Projects domain - data access layer.""" from __future__ import annotations import json from typing import Optional from uuid import UUID, uuid4 from sqlalchemy import func, select, text, update as sa_update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from src.core.ref_tokens import generate_public_ref from src.domain.projects.job_outbox_constants import ( JOB_KIND_CLOUD_SYNC_INITIAL, JOB_KIND_CLOUD_SYNC_POST_READY, JOB_KIND_PROJECT_SETUP, JOB_OUTBOX_MAX_ATTEMPTS, JOB_STATUS_FAILED, JOB_STATUS_PENDING, JOB_STATUS_PUBLISHED, JOB_STATUS_PUBLISHING, ) from src.domain.projects.models import ( ALLOWED_PROJECT_TYPES, MVP_PROJECT_TYPES, Project, ProjectEnvVar, ProjectImportSourceCatalog, ProjectImportSession, ProjectJobOutbox, ProjectRepo, ProjectSettings, ) # Cursor Agent CLI: persisted chat id for ``agent --resume`` (see cursor_cli / codegen worker). CURSOR_AGENT_SESSION_ID_KEY = "cursor_agent_session_id" DEFAULT_FORM_POLICY = { "country_mode": "all", "countries": [], "default_country": None, "require_subdivision": False, "phone_country_selector": True, } class ProjectImportSourceCatalogRepository: """Data access for Core-owned import-source catalog entries.""" def __init__(self, db: AsyncSession): self._db = db async def list_visible(self) -> list[ProjectImportSourceCatalog]: result = await self._db.execute( select(ProjectImportSourceCatalog) .where(ProjectImportSourceCatalog.visible.is_(True)) .order_by(ProjectImportSourceCatalog.sort_order.asc(), ProjectImportSourceCatalog.provider.asc()) ) return list(result.scalars().all()) async def upsert_defaults(self, defaults: list[dict]) -> None: for item in defaults: provider = str(item["provider"]) row = await self._db.get(ProjectImportSourceCatalog, provider) if row is None: row = ProjectImportSourceCatalog(provider=provider) self._db.add(row) row.label = str(item["label"]) row.icon = item.get("icon") row.sort_order = int(item.get("sort_order") or 100) row.visible = bool(item.get("visible", True)) row.metadata_json = dict(item.get("metadata_json") or {}) await self._db.flush() class ProjectRepository: """Project CRUD.""" def __init__(self, db: AsyncSession): self._db = db async def get_by_id(self, project_id: UUID) -> Optional[Project]: result = await self._db.execute(select(Project).where(Project.id == project_id)) return result.scalar_one_or_none() async def list_by_ids(self, project_ids: list[UUID]) -> list[Project]: if not project_ids: return [] result = await self._db.execute( select(Project).where(Project.id.in_(project_ids)).where(Project.deleted_at.is_(None)) ) return list(result.scalars().all()) async def get_by_project_ref(self, project_ref: str) -> Optional[Project]: """Lookup by stable routing ref (e.g. ``pabc12def``).""" ref = (project_ref or "").strip() if not ref: return None result = await self._db.execute(select(Project).where(Project.project_ref == ref)) return result.scalar_one_or_none() async def get_with_repo(self, project_id: UUID) -> Optional[Project]: """Get project with repo info. Attach repo to project for convenience.""" result = await self._db.execute( select(Project, ProjectRepo) .outerjoin(ProjectRepo, Project.id == ProjectRepo.project_id) .where(Project.id == project_id) ) row = result.one_or_none() if not row: return None project, repo = row project.repo = repo # type: ignore return project async def list_pages_projects_with_repos( self, project_ids: list[UUID] | None = None, ) -> list[Project]: """Active HTML/React/Nuxt projects that have a GitHub repo link.""" stmt = ( select(Project, ProjectRepo) .join(ProjectRepo, Project.id == ProjectRepo.project_id) .where(Project.deleted_at.is_(None)) .where(Project.type.in_(MVP_PROJECT_TYPES)) .order_by(Project.created_at.asc()) ) if project_ids: stmt = stmt.where(Project.id.in_(project_ids)) rows = (await self._db.execute(stmt)).all() found: list[Project] = [] for project, repo in rows: project.repo = repo # type: ignore found.append(project) return found async def slug_exists_in_workspace(self, workspace_id: UUID, slug: str) -> bool: result = await self._db.execute( select(Project.id).where(Project.workspace_id == workspace_id).where(Project.slug == slug) ) return result.scalar_one_or_none() is not None @property def session(self) -> AsyncSession: return self._db async def acquire_workspace_advisory_xact_lock(self, workspace_id: UUID) -> None: """Serialize quota/slug allocation per workspace until transaction ends.""" b = workspace_id.bytes k1 = int.from_bytes(b[0:4], "big") & 0x7FFFFFFF k2 = int.from_bytes(b[4:8], "big") & 0x7FFFFFFF await self._db.execute(text("SELECT pg_advisory_xact_lock(:k1, :k2)"), {"k1": k1, "k2": k2}) async def get_stale_platform_files( self, current_version: str, limit: int = 50 ) -> list[tuple["Project", "ProjectRepo"]]: """Return (Project, ProjectRepo) pairs whose platform_files_version is NULL or behind.""" from sqlalchemy import or_ result = await self._db.execute( select(Project, ProjectRepo) .join(ProjectRepo, Project.id == ProjectRepo.project_id) .where(Project.deleted_at.is_(None)) .where(Project.status == "ready") .where( or_( Project.platform_files_version.is_(None), Project.platform_files_version != current_version, ) ) .limit(limit) ) return list(result.all()) async def set_platform_files_version(self, project_id: UUID, version: str) -> None: await self._db.execute( sa_update(Project) .where(Project.id == project_id) .values(platform_files_version=version) ) await self._db.flush() async def merge_setup_step(self, project_id: UUID, step_key: str, patch: dict) -> None: """Merge observability patch into ``projects.setup_steps[step_key]``.""" # One atomic UPDATE that touches only this step's key. A read-modify-write of the whole # dict lets a concurrent step's stale copy overwrite an entry (e.g. auth_provision). result = await self._db.execute( text( "UPDATE projects SET setup_steps = jsonb_set(" " COALESCE(setup_steps, '{}'::jsonb), ARRAY[:step_key]," " COALESCE(setup_steps -> :step_key, '{}'::jsonb) || CAST(:patch AS jsonb), true)" " WHERE id = :project_id" ), { "step_key": step_key, "patch": json.dumps(patch, default=str), "project_id": project_id, }, ) if result.rowcount == 0: return await self._db.flush() async def update_runtime_invariants(self, project_id: UUID, invariants: dict) -> None: """Persist the merged ``ProjectRuntimeInvariants`` dict. No-op if unchanged.""" p = await self.get_by_id(project_id) if not p or (p.runtime_invariants or {}) == invariants: return await self._db.execute( sa_update(Project).where(Project.id == project_id).values(runtime_invariants=invariants) ) await self._db.flush() async def create( self, workspace_id: UUID, slug: str, name: str, project_type: str, description: Optional[str] = None, created_by: Optional[UUID] = None, template_id: Optional[UUID] = None, template_ref: Optional[str] = None, template_github_repo: Optional[str] = None, template_starter_branch: Optional[str] = None, template_starter_sha: Optional[str] = None, source_kind: str = "generated", source_workspace_user_github_installation_id: Optional[UUID] = None, source_workspace_user_gitlab_connection_id: Optional[UUID] = None, source_workspace_user_bitbucket_connection_id: Optional[UUID] = None, source_workspace_user_azure_connection_id: Optional[UUID] = None, source_repo_full_name: Optional[str] = None, source_branch: Optional[str] = None, source_root_dir: Optional[str] = None, source_import_env: bool = True, project_id: Optional[UUID] = None, hosting_provider: Optional[str] = None, ) -> Project: if project_type not in ALLOWED_PROJECT_TYPES: raise ValueError(f"Invalid project type. Allowed: {ALLOWED_PROJECT_TYPES}") from src.domain.hosting.profile import provider_for_new_project if hosting_provider is None: hosting_provider = provider_for_new_project(project_type) pid = project_id or uuid4() github_repo_name = f"p2hs-{pid.hex}" if len(github_repo_name) > 100: github_repo_name = github_repo_name[:100] p = Project( id=pid, workspace_id=workspace_id, slug=slug, github_repo_name=github_repo_name, name=name, type=project_type, description=description, created_by=created_by, template_id=template_id, template_ref=template_ref, template_github_repo=template_github_repo, template_starter_branch=template_starter_branch, template_starter_sha=template_starter_sha, source_kind=source_kind, source_workspace_user_github_installation_id=source_workspace_user_github_installation_id, source_workspace_user_gitlab_connection_id=source_workspace_user_gitlab_connection_id, source_workspace_user_bitbucket_connection_id=source_workspace_user_bitbucket_connection_id, source_workspace_user_azure_connection_id=source_workspace_user_azure_connection_id, source_repo_full_name=source_repo_full_name, source_branch=source_branch, source_root_dir=source_root_dir, source_import_env=source_import_env, primary_hosting_provider=hosting_provider, status="creating", ) self._db.add(p) await self._db.flush() if not p.project_ref: for _ in range(40): cand = "p" + generate_public_ref(7, prefix="") taken = await self._db.execute( select(Project.id).where(Project.project_ref == cand).limit(1) ) if taken.scalar_one_or_none() is None: p.project_ref = cand break else: raise RuntimeError("Could not allocate unique project_ref") await self._db.flush() await self._db.refresh(p) return p async def list_for_workspace(self, workspace_id: UUID) -> list[Project]: result = await self._db.execute( select(Project).where(Project.workspace_id == workspace_id).where(Project.deleted_at.is_(None)) ) return list(result.scalars().all()) async def update( self, project_id: UUID, name: Optional[str] = None, description: Optional[str] = None, status: Optional[str] = None, active_revision_id: Optional[UUID] = None, active_preview_id: Optional[UUID] = None, active_session_id: Optional[UUID] = None, boilerplate_version: Optional[str] = None, source_imported_commit_sha: Optional[str] = None, ) -> Optional[Project]: p = await self.get_by_id(project_id) if not p: return None if name is not None: p.name = name if description is not None: p.description = description if status is not None: p.status = status if active_revision_id is not None: p.active_revision_id = active_revision_id if active_preview_id is not None: p.active_preview_id = active_preview_id if active_session_id is not None: p.active_session_id = active_session_id if boilerplate_version is not None: p.boilerplate_version = boilerplate_version if source_imported_commit_sha is not None: p.source_imported_commit_sha = source_imported_commit_sha await self._db.flush() return p async def update_hostnames( self, project_id: UUID, *, live_hostname: Optional[str] = None, preview_hostname: Optional[str] = None, ) -> Optional[Project]: p = await self.get_by_id(project_id) if not p: return None if live_hostname is not None: p.live_hostname = live_hostname if preview_hostname is not None: p.preview_hostname = preview_hostname await self._db.flush() return p async def soft_delete(self, project_id: UUID) -> None: from datetime import datetime, timezone p = await self.get_by_id(project_id) if p: p.deleted_at = datetime.now(timezone.utc) await self._db.flush() class ProjectRepoRepository: """Project repo CRUD.""" def __init__(self, db: AsyncSession): self._db = db async def get_by_project_id(self, project_id: UUID) -> Optional[ProjectRepo]: result = await self._db.execute(select(ProjectRepo).where(ProjectRepo.project_id == project_id)) return result.scalar_one_or_none() async def create(self, project_id: UUID, repo_full_name: str, github_repo_id: Optional[int] = None) -> ProjectRepo: r = ProjectRepo(project_id=project_id, repo_full_name=repo_full_name, github_repo_id=github_repo_id) self._db.add(r) await self._db.flush() await self._db.refresh(r) return r async def update_remote_metadata( self, project_id: UUID, *, repo_full_name: Optional[str] = None, github_repo_id: Optional[int] = None, clone_url: Optional[str] = None, html_url: Optional[str] = None, default_branch: Optional[str] = None, preview_branch: Optional[str] = None, ) -> Optional[ProjectRepo]: r = await self.get_by_project_id(project_id) if not r: return None if repo_full_name is not None: r.repo_full_name = repo_full_name if github_repo_id is not None: r.github_repo_id = github_repo_id if clone_url is not None: r.clone_url = clone_url if html_url is not None: r.html_url = html_url if default_branch is not None: r.default_branch = default_branch if preview_branch is not None: r.preview_branch = preview_branch await self._db.flush() return r async def update_setup_state( self, project_id: UUID, *, setup_stage: Optional[str] = None, initial_commit_sha: Optional[str] = None, setup_preview_id: Optional[UUID] = None, ) -> Optional[ProjectRepo]: r = await self.get_by_project_id(project_id) if not r: return None if setup_stage is not None: r.setup_stage = setup_stage if initial_commit_sha is not None: r.initial_commit_sha = initial_commit_sha if setup_preview_id is not None: r.setup_preview_id = setup_preview_id await self._db.flush() return r class ProjectImportSessionRepository: """Import start/finalize handshake state.""" def __init__(self, db: AsyncSession): self._db = db async def create( self, *, workspace_id: UUID, created_by: UUID, provider: str = "github", project_id: Optional[UUID] = None, ) -> ProjectImportSession: row = ProjectImportSession( project_id=project_id or uuid4(), workspace_id=workspace_id, created_by=created_by, provider=provider, status="started", ) self._db.add(row) await self._db.flush() await self._db.refresh(row) return row async def get(self, project_id: UUID) -> Optional[ProjectImportSession]: result = await self._db.execute( select(ProjectImportSession).where(ProjectImportSession.project_id == project_id) ) return result.scalar_one_or_none() async def update( self, project_id: UUID, *, status: Optional[str] = None, github_installation_id: Optional[int] = None, workspace_user_github_installation_id: Optional[UUID] = None, workspace_user_gitlab_connection_id: Optional[UUID] = None, workspace_user_bitbucket_connection_id: Optional[UUID] = None, workspace_user_azure_connection_id: Optional[UUID] = None, source_repo_full_name: Optional[str] = None, source_branch: Optional[str] = None, source_root_dir: Optional[str] = None, ) -> Optional[ProjectImportSession]: row = await self.get(project_id) if not row: return None if status is not None: row.status = status if github_installation_id is not None: row.github_installation_id = github_installation_id if workspace_user_github_installation_id is not None: row.workspace_user_github_installation_id = workspace_user_github_installation_id if workspace_user_gitlab_connection_id is not None: row.workspace_user_gitlab_connection_id = workspace_user_gitlab_connection_id if workspace_user_bitbucket_connection_id is not None: row.workspace_user_bitbucket_connection_id = workspace_user_bitbucket_connection_id if workspace_user_azure_connection_id is not None: row.workspace_user_azure_connection_id = workspace_user_azure_connection_id if source_repo_full_name is not None: row.source_repo_full_name = source_repo_full_name if source_branch is not None: row.source_branch = source_branch if source_root_dir is not None: row.source_root_dir = source_root_dir await self._db.flush() return row class ProjectSettingsRepository: """Project settings CRUD.""" def __init__(self, db: AsyncSession): self._db = db async def get_or_create(self, project_id: UUID) -> ProjectSettings: result = await self._db.execute(select(ProjectSettings).where(ProjectSettings.project_id == project_id)) s = result.scalar_one_or_none() if not s: s = ProjectSettings(project_id=project_id) self._db.add(s) await self._db.flush() await self._db.refresh(s) return s async def get_persona(self, project_id: UUID) -> dict: s = await self.get_or_create(project_id) return (s.settings or {}).get("persona") or {} async def get_form_policy(self, project_id: UUID) -> dict: """Get the non-secret form-control policy consumed by codegen and the dashboard.""" s = await self.get_or_create(project_id) return {**DEFAULT_FORM_POLICY, **((s.settings or {}).get("form_policy") or {})} async def set_form_policy(self, project_id: UUID, policy: dict) -> None: """Persist project-scoped form controls without involving runtime environment variables.""" from sqlalchemy import update as sa_update s = await self.get_or_create(project_id) merged = {**(s.settings or {}), "form_policy": policy} await self._db.execute( sa_update(ProjectSettings) .where(ProjectSettings.project_id == project_id) .values(settings=merged) ) await self._db.flush() async def set_persona(self, project_id: UUID, persona: dict) -> None: s = await self.get_or_create(project_id) updated = {**(s.settings or {}), "persona": persona} from sqlalchemy import update as sa_update await self._db.execute( sa_update(ProjectSettings) .where(ProjectSettings.project_id == project_id) .values(settings=updated) ) async def get_search_index_adapters(self, project_id: UUID) -> list[dict]: settings = await self.get_or_create(project_id) value = (settings.settings or {}).get("search_index_adapters") or [] return [item for item in value if isinstance(item, dict)] async def set_search_index_adapter(self, project_id: UUID, adapter: dict) -> list[dict]: from sqlalchemy import update as sa_update settings = await self.get_or_create(project_id) existing = await self.get_search_index_adapters(project_id) replaced = [item for item in existing if item.get("entity_key") != adapter.get("entity_key")] replaced.append(adapter) merged = {**(settings.settings or {}), "search_index_adapters": replaced} await self._db.execute( sa_update(ProjectSettings).where(ProjectSettings.project_id == project_id).values(settings=merged) ) await self._db.flush() return replaced async def delete_search_index_adapter(self, project_id: UUID, entity_key: str) -> list[dict]: from sqlalchemy import update as sa_update settings = await self.get_or_create(project_id) remaining = [item for item in await self.get_search_index_adapters(project_id) if item.get("entity_key") != entity_key] merged = {**(settings.settings or {}), "search_index_adapters": remaining} await self._db.execute( sa_update(ProjectSettings).where(ProjectSettings.project_id == project_id).values(settings=merged) ) await self._db.flush() return remaining async def merge_builder_user_values( self, project_id: UUID, values: dict[str, str], *, overwrite: bool = True ) -> None: """Merge owner-supplied public builder keys (Calendly URL, etc.) into ``settings.builder_user_values``.""" from sqlalchemy import update as sa_update s = await self.get_or_create(project_id) merged = {**(s.settings or {})} existing = {**(merged.get("builder_user_values") or {})} if overwrite: buv = {**existing, **values} else: buv = {**values, **existing} merged["builder_user_values"] = buv await self._db.execute( sa_update(ProjectSettings) .where(ProjectSettings.project_id == project_id) .values(settings=merged) ) await self._db.flush() async def get_cursor_agent_session_id(self, project_id: UUID) -> Optional[str]: s = await self.get_or_create(project_id) raw = (s.settings or {}).get(CURSOR_AGENT_SESSION_ID_KEY) if isinstance(raw, str) and raw.strip(): return raw.strip() return None async def set_cursor_agent_session_id(self, project_id: UUID, session_id: str) -> None: if not session_id or not str(session_id).strip(): return from sqlalchemy import update as sa_update s = await self.get_or_create(project_id) merged = {**(s.settings or {}), CURSOR_AGENT_SESSION_ID_KEY: str(session_id).strip()} await self._db.execute( sa_update(ProjectSettings) .where(ProjectSettings.project_id == project_id) .values(settings=merged) ) await self._db.flush() async def clear_cursor_agent_session_id(self, project_id: UUID) -> None: """Remove stored Cursor agent chat id so the next run starts a fresh agent session.""" from sqlalchemy import update as sa_update s = await self.get_or_create(project_id) merged = {**(s.settings or {})} merged.pop(CURSOR_AGENT_SESSION_ID_KEY, None) await self._db.execute( sa_update(ProjectSettings) .where(ProjectSettings.project_id == project_id) .values(settings=merged) ) await self._db.flush() class ProjectEnvVarRepository: """Project env vars CRUD.""" def __init__(self, db: AsyncSession): self._db = db async def create( self, project_id: UUID, key: str, encrypted_value: str, environment: str = "all", visibility: str = "secret", ) -> ProjectEnvVar: ev = ProjectEnvVar( project_id=project_id, key=key, encrypted_value=encrypted_value, environment=environment, visibility=visibility, ) self._db.add(ev) await self._db.flush() await self._db.refresh(ev) return ev async def list_for_project(self, project_id: UUID, environment: Optional[str] = None) -> list[ProjectEnvVar]: q = select(ProjectEnvVar).where(ProjectEnvVar.project_id == project_id) if environment: q = q.where(ProjectEnvVar.environment.in_((environment, "all"))) result = await self._db.execute(q) return list(result.scalars().all()) async def list_all_for_deploy(self, project_id: UUID) -> list[ProjectEnvVar]: """All rows regardless of environment — used by propagation adapter.""" result = await self._db.execute( select(ProjectEnvVar).where(ProjectEnvVar.project_id == project_id) ) return list(result.scalars().all()) async def get_by_id(self, project_id: UUID, env_var_id: UUID) -> Optional[ProjectEnvVar]: result = await self._db.execute( select(ProjectEnvVar).where( ProjectEnvVar.project_id == project_id, ProjectEnvVar.id == env_var_id, ) ) return result.scalar_one_or_none() async def get_by_key( self, project_id: UUID, key: str, environment: str = "all" ) -> Optional[ProjectEnvVar]: result = await self._db.execute( select(ProjectEnvVar).where( ProjectEnvVar.project_id == project_id, ProjectEnvVar.key == key, ProjectEnvVar.environment == environment, ) ) return result.scalar_one_or_none() async def upsert_by_key( self, project_id: UUID, key: str, encrypted_value: str, environment: str = "all", visibility: str = "secret", ) -> ProjectEnvVar: """Create or overwrite an env var row identified by (project_id, key, environment).""" from sqlalchemy import update as sa_update existing = await self.get_by_key(project_id, key, environment) if existing: await self._db.execute( sa_update(ProjectEnvVar) .where(ProjectEnvVar.id == existing.id) .values(encrypted_value=encrypted_value, visibility=visibility) ) await self._db.flush() await self._db.refresh(existing) return existing return await self.create( project_id=project_id, key=key, encrypted_value=encrypted_value, environment=environment, visibility=visibility, ) async def update_encrypted_value( self, env_var_id: UUID, encrypted_value: str, visibility: Optional[str] = None ) -> Optional[ProjectEnvVar]: from sqlalchemy import update as sa_update values: dict = {"encrypted_value": encrypted_value} if visibility is not None: values["visibility"] = visibility await self._db.execute( sa_update(ProjectEnvVar) .where(ProjectEnvVar.id == env_var_id) .values(**values) ) await self._db.flush() result = await self._db.execute(select(ProjectEnvVar).where(ProjectEnvVar.id == env_var_id)) return result.scalar_one_or_none() async def delete_by_id(self, project_id: UUID, env_var_id: UUID) -> bool: from sqlalchemy import delete as sa_delete res = await self._db.execute( sa_delete(ProjectEnvVar).where( ProjectEnvVar.project_id == project_id, ProjectEnvVar.id == env_var_id, ) ) await self._db.flush() return res.rowcount > 0 class ProjectJobOutboxRepository: """Transactional outbox: pending rows committed with project creation.""" def __init__(self, db: AsyncSession): self._db = db async def insert_pending(self, project_id: UUID, job_kind: str, payload: Optional[dict] = None) -> bool: """Return True if a new pending row was inserted (SAVEPOINT rolled back cleanly on duplicate).""" try: async with self._db.begin_nested(): row = ProjectJobOutbox( project_id=project_id, job_kind=job_kind, payload_json=payload or {}, status=JOB_STATUS_PENDING, ) self._db.add(row) await self._db.flush() except IntegrityError: return False return True async def insert_initial_project_jobs(self, project_id: UUID) -> None: """Enqueue intents for GitHub setup then Cloud bootstrap (dispatcher publishes order).""" await self.insert_pending(project_id, JOB_KIND_PROJECT_SETUP, {}) await self.insert_pending(project_id, JOB_KIND_CLOUD_SYNC_INITIAL, {}) async def reset_stale_publishing_locks(self, *, stale_seconds: int) -> int: from datetime import datetime, timedelta, timezone cutoff = datetime.now(timezone.utc) - timedelta(seconds=stale_seconds) res = await self._db.execute( sa_update(ProjectJobOutbox) .where(ProjectJobOutbox.status == JOB_STATUS_PUBLISHING) .where(ProjectJobOutbox.publishing_started_at.is_not(None)) .where(ProjectJobOutbox.publishing_started_at < cutoff) .values( status=JOB_STATUS_PENDING, publishing_started_at=None, last_error="stale_publishing_reset", ) ) await self._db.flush() return res.rowcount or 0 async def claim_batch_for_publishing(self, limit: int = 25) -> list[ProjectJobOutbox]: """Atomically move up to ``limit`` pending rows to publishing (SKIP LOCKED).""" sql = text( """ WITH picked AS ( SELECT id FROM project_job_outbox WHERE status = :pending ORDER BY CASE job_kind WHEN :jk_setup THEN 0 WHEN :jk_cloud_initial THEN 1 WHEN :jk_cloud_post THEN 2 ELSE 9 END, created_at ASC LIMIT :lim FOR UPDATE SKIP LOCKED ) UPDATE project_job_outbox AS j SET status = :publishing, publishing_started_at = NOW() FROM picked WHERE j.id = picked.id RETURNING j.* """ ) result = await self._db.execute( sql, { "pending": JOB_STATUS_PENDING, "publishing": JOB_STATUS_PUBLISHING, "jk_setup": JOB_KIND_PROJECT_SETUP, "jk_cloud_initial": JOB_KIND_CLOUD_SYNC_INITIAL, "jk_cloud_post": JOB_KIND_CLOUD_SYNC_POST_READY, "lim": limit, }, ) rows = result.mappings().all() out: list[ProjectJobOutbox] = [] for m in rows: ob = ProjectJobOutbox( id=m["id"], project_id=m["project_id"], job_kind=m["job_kind"], payload_json=m["payload_json"] or {}, status=m["status"], attempts=m["attempts"], last_error=m["last_error"], publishing_started_at=m["publishing_started_at"], created_at=m["created_at"], published_at=m["published_at"], ) out.append(ob) return out async def mark_published(self, row_id: UUID) -> None: await self._db.execute( sa_update(ProjectJobOutbox) .where(ProjectJobOutbox.id == row_id) .values( status=JOB_STATUS_PUBLISHED, published_at=func.now(), publishing_started_at=None, last_error=None, ) ) await self._db.flush() async def mark_enqueue_failed(self, row_id: UUID, error_message: str) -> None: row = ( await self._db.execute(select(ProjectJobOutbox).where(ProjectJobOutbox.id == row_id)) ).scalar_one_or_none() if not row: return attempts = int(row.attempts or 0) + 1 if attempts >= JOB_OUTBOX_MAX_ATTEMPTS: await self._db.execute( sa_update(ProjectJobOutbox) .where(ProjectJobOutbox.id == row_id) .values( status=JOB_STATUS_FAILED, attempts=attempts, last_error=(error_message or "")[:8000], publishing_started_at=None, ) ) else: await self._db.execute( sa_update(ProjectJobOutbox) .where(ProjectJobOutbox.id == row_id) .values( status=JOB_STATUS_PENDING, attempts=attempts, last_error=(error_message or "")[:8000], publishing_started_at=None, ) ) await self._db.flush()