From 7a3c2452e1e0625e7a7fdc710be842313b330ce1 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Mon, 3 Aug 2026 10:51:55 +0800 Subject: [PATCH] feat: auth --- backend/src/backend/auth.py | 256 +++++++++++++++++ common/src/common/auth/__init__.py | 14 + common/src/common/auth/jwt.py | 141 ++++++++++ common/src/common/auth/membership.py | 69 +++++ common/src/common/auth/passwords.py | 55 ++++ common/src/common/scheduler/trigger.py | 370 +++++++++++++++++++++++++ 6 files changed, 905 insertions(+) create mode 100644 backend/src/backend/auth.py create mode 100644 common/src/common/auth/__init__.py create mode 100644 common/src/common/auth/jwt.py create mode 100644 common/src/common/auth/membership.py create mode 100644 common/src/common/auth/passwords.py create mode 100644 common/src/common/scheduler/trigger.py diff --git a/backend/src/backend/auth.py b/backend/src/backend/auth.py new file mode 100644 index 0000000..9fb1928 --- /dev/null +++ b/backend/src/backend/auth.py @@ -0,0 +1,256 @@ +"""Cookie+JWT authentication endpoints. + +The user-facing flow is: + 1. POST /api/v1/auth/login — verify password, set HttpOnly cookie + 2. every other /api/ request reads the cookie via + ``backend.dependencies.request_context`` + 3. POST /api/v1/auth/logout — clear the cookie + 4. GET /api/v1/auth/me — return the current user + +Service-to-service calls do not use these endpoints — they live on the +shared Docker network and have no application-layer auth. See +``docker-compose.yml`` and ``schedule/service.py``. +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, Request, Response, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.dependencies import database_session +from common.auth.jwt import JwtError, issue_jwt, verify_jwt_token +from common.auth.passwords import verify_password +from common.db.models import Roles, Users, WorkspaceMembers, Workspaces +from common.ids import new_ulid + + +router = APIRouter(tags=["auth"]) + +# Cookie config. ``secure=True`` requires HTTPS — the only safe +# assumption in production. Dev environments running on plain HTTP +# should reverse-proxy with TLS termination or set the env knob +# (future extension). +COOKIE_NAME = "access_token" +COOKIE_TTL_SECONDS = 24 * 60 * 60 +COOKIE_SECURE = True +COOKIE_SAMESITE = "lax" + + +def _set_session_cookie(response: Response, token: str) -> None: + response.set_cookie( + key=COOKIE_NAME, + value=token, + max_age=COOKIE_TTL_SECONDS, + path="/", + httponly=True, + secure=COOKIE_SECURE, + samesite=COOKIE_SAMESITE, + ) + + +def _clear_session_cookie(response: Response) -> None: + response.delete_cookie(key=COOKIE_NAME, path="/") + + +def _user_payload(user: Users, role_code: str | None = None) -> dict[str, Any]: + return { + "user_id": user.user_id, + "username": user.username, + "display_name": user.display_name, + "email": user.email, + "status": user.status, + "role_code": role_code, + } + + +def _workspace_payload( + workspace: Workspaces, + role: Roles, +) -> dict[str, Any]: + return { + "workspace_id": workspace.workspace_id, + "workspace_code": workspace.workspace_code, + "workspace_name": workspace.workspace_name, + "role_code": role.role_code, + "role_name": role.role_name, + } + + +@router.post("/api/v1/auth/login") +async def login( + request: Request, + response: Response, + session: AsyncSession = Depends(database_session), +) -> dict[str, Any]: + """Verify username/password and issue a session cookie. + + Returns the user record and the workspaces they are an active + member of (joined earliest first, which doubles as the default + selection until the user picks a different one in the UI). The + workspace list is informational — the JWT itself does not bind a + workspace; each request specifies its own ``?workspace_id=``. + """ + body = await request.json() + username = (body or {}).get("username", "").strip() + password = (body or {}).get("password", "") + if not username or not password: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + "username and password are required", + ) + + user = await session.scalar( + select(Users).where( + Users.username == username, + Users.status == "active", + ) + ) + if user is None or not verify_password(password, user.password_hash): + # Unified 401 to prevent username enumeration. + raise HTTPException( + status.HTTP_401_UNAUTHORIZED, + "invalid username or password", + ) + + # Pull all active memberships. The earliest join wins as default + # because there is no `is_default` column on `workspace_members`. + rows = ( + await session.execute( + select(Workspaces, Roles, WorkspaceMembers.joined_at) + .join( + WorkspaceMembers, + WorkspaceMembers.workspace_id == Workspaces.workspace_id, + ) + .join(Roles, Roles.role_id == WorkspaceMembers.role_id) + .where( + WorkspaceMembers.user_id == user.user_id, + WorkspaceMembers.member_status == "active", + Workspaces.status == "active", + ) + .order_by(WorkspaceMembers.joined_at.asc()) + ) + ).all() + if not rows: + raise HTTPException( + status.HTTP_403_FORBIDDEN, + "user is not a member of any active workspace", + ) + + workspaces = [] + default_workspace_id: str | None = None + for workspace, role, joined_at in rows: + workspaces.append(_workspace_payload(workspace, role)) + if default_workspace_id is None: + default_workspace_id = workspace.workspace_id + + # Pick a default role_code for the user payload: prefer admin if + # the user has it in any workspace, otherwise use the first one + # returned. This is only for UI greeting; access control checks + # run on a per-request basis via the chosen workspace_id. + user_role_code: str | None = None + for workspace, role, _ in rows: + if role.role_code == "admin": + user_role_code = "admin" + break + if user_role_code is None: + user_role_code = rows[0][1].role_code + + token = issue_jwt(user.user_id, ttl_seconds=COOKIE_TTL_SECONDS) + _set_session_cookie(response, token) + + return { + "request_id": new_ulid(), + "data": { + "user": _user_payload(user, user_role_code), + "workspaces": workspaces, + "default_workspace_id": default_workspace_id, + }, + "meta": {}, + } + + +@router.post("/api/v1/auth/logout") +async def logout(response: Response) -> dict[str, Any]: + """Clear the session cookie. Idempotent.""" + _clear_session_cookie(response) + return { + "request_id": new_ulid(), + "data": {"logged_out": True}, + "meta": {}, + } + + +@router.get("/api/v1/auth/me") +async def me( + request: Request, + session: AsyncSession = Depends(database_session), +) -> dict[str, Any]: + """Return the current user record based on the session cookie. + + Does not require a workspace_id — useful for the frontend to + bootstrap identity on app load before any workspace has been + selected. Workspace list is included so the login screen can be + skipped on subsequent visits. + """ + token = request.cookies.get(COOKIE_NAME) + if not token: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not authenticated") + try: + payload = verify_jwt_token(token) + except JwtError as exc: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(exc)) from exc + + user_id = payload.get("sub") + if not user_id: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token") + + user = await session.get(Users, user_id) + if user is None or user.status != "active": + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found") + + rows = ( + await session.execute( + select(Workspaces, Roles, WorkspaceMembers.joined_at) + .join( + WorkspaceMembers, + WorkspaceMembers.workspace_id == Workspaces.workspace_id, + ) + .join(Roles, Roles.role_id == WorkspaceMembers.role_id) + .where( + WorkspaceMembers.user_id == user.user_id, + WorkspaceMembers.member_status == "active", + Workspaces.status == "active", + ) + .order_by(WorkspaceMembers.joined_at.asc()) + ) + ).all() + + workspaces = [_workspace_payload(ws, role) for ws, role, _ in rows] + default_workspace_id = workspaces[0]["workspace_id"] if workspaces else None + user_role_code: str | None = None + for _ws, role, _ in rows: + if role.role_code == "admin": + user_role_code = "admin" + break + if user_role_code is None and rows: + user_role_code = rows[0][1].role_code + + return { + "request_id": new_ulid(), + "data": { + "user": _user_payload(user, user_role_code), + "workspaces": workspaces, + "default_workspace_id": default_workspace_id, + }, + "meta": {}, + } + + +__all__ = [ + "COOKIE_NAME", + "COOKIE_TTL_SECONDS", + "router", +] diff --git a/common/src/common/auth/__init__.py b/common/src/common/auth/__init__.py new file mode 100644 index 0000000..7de60de --- /dev/null +++ b/common/src/common/auth/__init__.py @@ -0,0 +1,14 @@ +"""Authentication primitives shared across services. + +Houses the HS256 JWT issuer/verifier, bcrypt password helpers, and the +single canonical active-membership loader. The previous implementation +lived inline inside ``backend/jupyter.py`` and ``backend/dependencies.py``; +moving it here lets the schedule service verify the same tokens (when +service-to-service auth is reintroduced) and keeps the dependency +inversion clean. + +Service-to-service HTTP calls in this repository do NOT currently +authenticate at the application layer (see ``docker-compose.yml``: only +the gateway exposes a host port). The JWT and membership helpers are +used exclusively by user-facing endpoints. +""" diff --git a/common/src/common/auth/jwt.py b/common/src/common/auth/jwt.py new file mode 100644 index 0000000..9be20ef --- /dev/null +++ b/common/src/common/auth/jwt.py @@ -0,0 +1,141 @@ +"""HS256 JWT issuance and verification. + +The implementation is intentionally minimal: a hand-rolled HS256 +signer/verifier so the project does not depend on PyJWT. It deliberately +ignores the ``alg`` header on the verify side and always recomputes +HMAC-SHA256, which means a forged ``"alg":"none"`` token still fails +signature validation. + +Token payload contract: + sub - user_id (CHAR(26) ULID) + exp - unix seconds; mandatory + iat - unix seconds; mandatory (used for last_logout_at checks if added later) + +Any other claim is preserved by the verifier but not interpreted here. +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import time +from typing import Any + +from common.config import settings + + +JWT_SECRET: str = settings.jwt_secret +JWT_ALGORITHM: str = "HS256" +DEFAULT_TTL_SECONDS: int = 24 * 60 * 60 + + +def _b64encode(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def _b64decode(value: str) -> bytes: + padding = "=" * (-len(value) % 4) + return base64.urlsafe_b64decode(value + padding) + + +def issue_jwt( + user_id: str, + *, + ttl_seconds: int = DEFAULT_TTL_SECONDS, + extra_claims: dict[str, Any] | None = None, + now: int | None = None, +) -> str: + """Sign a JWT for ``user_id`` and return the compact serialization. + + The header is fixed ``{"alg":"HS256","typ":"JWT"}``; the signature + uses HMAC-SHA256 over ``{header_b64}.{payload_b64}`` keyed by + ``settings.jwt_secret``. + + ``ttl_seconds`` defaults to 24h. ``extra_claims`` is merged into the + payload after ``sub``/``iat``/``exp`` are populated and would + overwrite those if callers passed the same keys — kept simple on + purpose so we never accidentally bypass the contract. + """ + if not user_id: + raise ValueError("user_id is required") + + issued_at = int(time.time()) if now is None else int(now) + payload: dict[str, Any] = { + "sub": user_id, + "iat": issued_at, + "exp": issued_at + int(ttl_seconds), + } + if extra_claims: + payload.update(extra_claims) + + header = {"alg": JWT_ALGORITHM, "typ": "JWT"} + header_b64 = _b64encode(json.dumps(header, separators=(",", ":")).encode()) + payload_b64 = _b64encode(json.dumps(payload, separators=(",", ":")).encode()) + signing_input = f"{header_b64}.{payload_b64}".encode() + signature = hmac.new( + JWT_SECRET.encode(), + signing_input, + hashlib.sha256, + ).digest() + signature_b64 = _b64encode(signature) + return f"{header_b64}.{payload_b64}.{signature_b64}" + + +class JwtError(Exception): + """Raised on missing / malformed / expired / wrong-signature tokens.""" + + +def verify_jwt_token(token: str | None) -> dict[str, Any]: + """Verify an HS256-signed JWT and return its payload. + + The function is the inverse of :func:`issue_jwt`. The ``alg`` header + is read for completeness but the signature is always recomputed + under HS256 — a token claiming ``alg":"none"`` is rejected because + its signature segment will not match a recomputed HMAC. + + Raises :class:`JwtError` on every failure mode; callers translate to + 401 in HTTP contexts. + """ + if not token: + raise JwtError("missing authentication token") + + try: + header_b64, payload_b64, signature_b64 = token.split(".", 2) + except ValueError as exc: + raise JwtError("malformed token") from exc + + signing_input = f"{header_b64}.{payload_b64}".encode() + expected = hmac.new( + JWT_SECRET.encode(), + signing_input, + hashlib.sha256, + ).digest() + try: + signature = _b64decode(signature_b64) + except Exception as exc: # pragma: no cover - malformed b64 + raise JwtError("malformed signature") from exc + if not hmac.compare_digest(expected, signature): + raise JwtError("signature mismatch") + + try: + payload = json.loads(_b64decode(payload_b64)) + except Exception as exc: + raise JwtError("malformed payload") from exc + + exp = payload.get("exp") + if not isinstance(exp, (int, float)) or exp < time.time(): + raise JwtError("token expired") + + return payload + + +__all__ = [ + "DEFAULT_TTL_SECONDS", + "JWT_ALGORITHM", + "JWT_SECRET", + "JwtError", + "issue_jwt", + "verify_jwt_token", +] diff --git a/common/src/common/auth/membership.py b/common/src/common/auth/membership.py new file mode 100644 index 0000000..b6e528d --- /dev/null +++ b/common/src/common/auth/membership.py @@ -0,0 +1,69 @@ +"""Active workspace-membership loader. + +Single canonical implementation that verifies a user is currently a +member of the given workspace. Used by the dependency-injection layer +in the backend; ``jupyter.py`` previously carried its own near-duplicate +that is now reduced to a thin caller. + +The query joins Users / WorkspaceMembers / Workspaces / Roles so +callers receive the role they need for ``is_admin``-style checks +without a second round-trip. +""" + +from __future__ import annotations + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from common.db.models import Roles, Users, WorkspaceMembers, Workspaces + + +class MembershipError(Exception): + """Raised when the user is not an active member of the workspace.""" + + +async def load_active_membership( + session: AsyncSession, + user_id: str, + workspace_id: str, +) -> tuple[Users, Workspaces, Roles]: + """Return ``(user, workspace, role)`` for an active membership. + + All four conditions must hold: ``Users.status == 'active'``, + ``WorkspaceMembers.member_status == 'active'``, + ``Workspaces.status == 'active'``, and the row exists at all. + Raises :class:`MembershipError` otherwise. + """ + statement = ( + select(Users, Workspaces, Roles) + .join( + WorkspaceMembers, + WorkspaceMembers.user_id == Users.user_id, + ) + .join( + Workspaces, + Workspaces.workspace_id == WorkspaceMembers.workspace_id, + ) + .join( + Roles, + Roles.role_id == WorkspaceMembers.role_id, + ) + .where( + Users.user_id == user_id, + Users.status == "active", + WorkspaceMembers.workspace_id == workspace_id, + WorkspaceMembers.member_status == "active", + Workspaces.status == "active", + ) + ) + row = (await session.execute(statement)).one_or_none() + if row is None: + raise MembershipError("active workspace membership is required") + user, workspace, role = row + return user, workspace, role + + +__all__ = [ + "MembershipError", + "load_active_membership", +] diff --git a/common/src/common/auth/passwords.py b/common/src/common/auth/passwords.py new file mode 100644 index 0000000..f1e096f --- /dev/null +++ b/common/src/common/auth/passwords.py @@ -0,0 +1,55 @@ +"""Bcrypt password hashing helpers. + +Uses passlib's :class:`CryptContext` so the algorithm choice stays in +one place — when (not if) we move to argon2 we change the ``schemes`` +list and existing hashes still verify. + +The :data:`make_unusable_password` helper returns a bcrypt hash of a +random 32-byte secret. It is intentionally verifiable (to keep the +``verify_password`` path symmetric) but cannot be matched by any +human-supplied plaintext, so it is safe to assign to service accounts +that should never log in interactively. +""" + +from __future__ import annotations + +import secrets + +from passlib.context import CryptContext + + +_crypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto") + + +def hash_password(plain: str) -> str: + """Hash ``plain`` with bcrypt and return the encoded digest.""" + if not plain: + raise ValueError("plain must be a non-empty string") + return _crypt_context.hash(plain) + + +def verify_password(plain: str, hashed: str) -> bool: + """Return True iff ``plain`` matches ``hashed`` under the active scheme.""" + if not plain or not hashed: + return False + try: + return _crypt_context.verify(plain, hashed) + except ValueError: + return False + + +def make_unusable_password() -> str: + """Return a bcrypt hash of a random 32-byte secret. + + Service accounts store this so the ``Users.password_hash`` column is + populated and any stray login attempt is rejected by the password + check (random secret → impossible to brute force offline). + """ + return _crypt_context.hash(secrets.token_urlsafe(32)) + + +__all__ = [ + "hash_password", + "make_unusable_password", + "verify_password", +] diff --git a/common/src/common/scheduler/trigger.py b/common/src/common/scheduler/trigger.py new file mode 100644 index 0000000..6347010 --- /dev/null +++ b/common/src/common/scheduler/trigger.py @@ -0,0 +1,370 @@ +"""Shared schedule-trigger logic. + +Both the user-facing manual run endpoint (``backend.schedule_runs``) +and the schedule service's cron tick handler call into this module to +materialize a ``ScheduleRuns`` row plus the corresponding +``schedule.run.requested`` outbox event. The outbox is the single +source of truth for run dispatch — the schedule executor polls MySQL +and picks the row up. + +The schedule executor does NOT take any application-layer auth from +this codebase. Service-to-service calls on the shared Docker network +are intentionally unauthenticated; the ``triggered_by`` field stores +the user_id that originated the run (a human for manual runs, the +fixed ``_system_cron`` user for cron ticks) and the executor +re-verifies that user against ``Users.status='active'`` before doing +work. +""" + +from __future__ import annotations + +import hashlib +from typing import Any, Literal + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from common.db.models import ( + ScheduleEdges, + ScheduleNodes, + ScheduleRuns, + Schedules, + Scripts, + Versions, +) +from common.eventing import add_outbox_event, utcnow +from common.ids import new_ulid + + +# A stable user_id used for cron-triggered runs. The corresponding +# ``Users`` row is seeded by the auth-bootstrap migration so any audit +# query joining on ``ScheduleRuns.triggered_by`` still resolves. +SYSTEM_CRON_USER_ID = "01HZZZZZZZZZZZZZZZZZZZZZZCR" + + +TriggerType = Literal["manual", "cron", "api"] + + +class TriggerError(Exception): + """Raised when a run cannot be created. Subclasses carry the + appropriate HTTP status when surfaced from the backend router.""" + + +class ScheduleNotFound(TriggerError): + pass + + +class InvalidDag(TriggerError): + def __init__(self, errors: list[dict[str, Any]]) -> None: + super().__init__("schedule must contain a valid non-empty DAG") + self.errors = errors + + +class DagTooLarge(TriggerError): + pass + + +class InvalidNodeArguments(TriggerError): + def __init__(self, message: str) -> None: + super().__init__(message) + self.message = message + + +def normalize_idempotency_key( + workspace_id: str, + schedule_id: str, + value: str, + *, + min_length: int = 8, +) -> str: + """SHA-256 the (workspace, schedule, header) triple and prefix a + version tag. Centralized so the schedule service and the backend + router produce the same key. + """ + normalized = value.strip() + if len(normalized) < min_length: + raise TriggerError( + f"Idempotency-Key must contain at least {min_length} characters" + ) + digest = hashlib.sha256( + f"{workspace_id}:{schedule_id}:{normalized}".encode("utf-8") + ).hexdigest() + return f"run:v1:{digest}" + + +def parse_node_arguments(value: dict[str, Any] | None) -> list[str]: + """Turn a node's ``arguments_json`` dict into a list of CLI args. + + Mirrors the backend's old ``_arguments`` helper but raises + :class:`InvalidNodeArguments` instead of an HTTP exception, so + the schedule service can use it without importing FastAPI. + """ + payload = value or {} + raw = payload.get("_args") + result: list[str] = [str(item) for item in raw] if isinstance(raw, list) else [] + for key, item in payload.items(): + if key == "_args": + continue + option = f"--{key.replace('_', '-')}" + if item is True: + result.append(option) + elif item is False or item is None: + continue + elif isinstance(item, list): + for list_item in item: + result.extend((option, str(list_item))) + elif isinstance(item, (str, int, float)): + result.extend((option, str(item))) + else: + raise InvalidNodeArguments( + f"node argument {key!r} must be a scalar or list" + ) + return result + + +async def _load_schedule( + session: AsyncSession, + schedule_id: str, + workspace_id: str, + *, + for_update: bool = False, +) -> Schedules: + statement = select(Schedules).where( + Schedules.schedule_id == schedule_id, + Schedules.workspace_id == workspace_id, + Schedules.deleted_at.is_(None), + ) + if for_update: + statement = statement.with_for_update() + item = await session.scalar(statement) + if item is None: + raise ScheduleNotFound("schedule not found") + return item + + +async def _load_graph( + session: AsyncSession, + schedule_id: str, +) -> tuple[ + list[tuple[ScheduleNodes, Versions, Scripts]], + list[ScheduleEdges], +]: + node_rows = ( + await session.execute( + select(ScheduleNodes, Versions, Scripts) + .join(Versions, Versions.versions_id == ScheduleNodes.versions_id) + .join(Scripts, Scripts.script_id == Versions.script_id) + .where(ScheduleNodes.schedule_id == schedule_id) + .order_by(ScheduleNodes.created_at, ScheduleNodes.node_key) + ) + ).all() + edges = list( + ( + await session.scalars( + select(ScheduleEdges) + .where(ScheduleEdges.schedule_id == schedule_id) + .order_by(ScheduleEdges.created_at, ScheduleEdges.edge_id) + ) + ).all() + ) + return list(node_rows), edges + + +def _validate_dag( + nodes: list[ScheduleNodes], + edges: list[ScheduleEdges], + *, + max_nodes: int = 100, + max_edges: int = 500, +) -> list[dict[str, Any]]: + errors: list[dict[str, Any]] = [] + if not nodes: + errors.append( + { + "code": "DAG_EMPTY", + "message": "schedule must contain at least one node", + } + ) + return errors + if len(nodes) > max_nodes or len(edges) > max_edges: + raise DagTooLarge( + f"schedule exceeds the v1 execution size limit " + f"({len(nodes)} nodes / {len(edges)} edges > {max_nodes}/{max_edges})" + ) + + # Cycle detection via Kahn's algorithm. + in_degree: dict[str, int] = {n.node_id: 0 for n in nodes} + adjacency: dict[str, list[str]] = {n.node_id: [] for n in nodes} + for edge in edges: + if edge.source_node_id not in in_degree or edge.target_node_id not in in_degree: + errors.append( + { + "code": "DAG_EDGE_REFERENCES_MISSING_NODE", + "message": f"edge {edge.edge_id} references unknown node", + "edge_id": edge.edge_id, + } + ) + continue + adjacency[edge.source_node_id].append(edge.target_node_id) + in_degree[edge.target_node_id] += 1 + queue = [nid for nid, d in in_degree.items() if d == 0] + ordered: list[str] = [] + while queue: + queue.sort() + current = queue.pop(0) + ordered.append(current) + for neighbor in adjacency[current]: + in_degree[neighbor] -= 1 + if in_degree[neighbor] == 0: + queue.append(neighbor) + if len(ordered) != len(nodes): + cycle_nodes = [nid for nid, d in in_degree.items() if d > 0] + errors.append( + { + "code": "DAG_CYCLE", + "message": "schedule graph contains a directed cycle", + "node_ids": cycle_nodes, + } + ) + return errors + + +def _build_snapshot( + schedule: Schedules, + node_rows: list[tuple[ScheduleNodes, Versions, Scripts]], + edges: list[ScheduleEdges], +) -> dict[str, Any]: + return { + "schedule_name": schedule.schedule_name, + "workflow_version": schedule.workflow_version, + "max_concurrency": schedule.max_concurrency, + "failure_policy": schedule.failure_policy, + "nodes": [ + { + "node_id": node.node_id, + "node_key": node.node_key, + "versions_id": version.versions_id, + "script_type": script.script_type, + "artifact_object_id": version.artifact_object_id, + "artifact_path": version.artifact_path, + "timeout_seconds": node.timeout_seconds, + "retry_count": node.retry_count, + "retry_interval_sec": node.retry_interval_sec, + "arguments": parse_node_arguments(node.arguments_json), + } + for node, version, script in node_rows + ], + "edges": [ + { + "source_node_id": edge.source_node_id, + "target_node_id": edge.target_node_id, + } + for edge in edges + ], + } + + +async def create_scheduled_run( + session: AsyncSession, + *, + schedule_id: str, + workspace_id: str, + triggered_by_user_id: str, + trigger_type: TriggerType, + idempotency_key: str, + trace_id: str | None = None, +) -> tuple[ScheduleRuns, bool]: + """Create a ``ScheduleRuns`` row + outbox event in this session. + + Returns ``(run, is_new)``. ``is_new=False`` means the + idempotency_key was already used and the existing run is returned + unchanged. ``triggered_by_user_id`` is stored as-is — for cron + triggers pass :data:`SYSTEM_CRON_USER_ID`. + + The caller owns the transaction: ``create_scheduled_run`` flushes + the new row to surface the unique-constraint violation on + ``idempotency_key`` deterministically, then leaves the commit to + the caller's session lifecycle. Both the backend's + ``request_context`` (which uses ``session_scope``) and the + schedule service's own session scope can wrap this call. + """ + # 1) Existing run short-circuit (re-using a known idempotency key). + existing = await session.scalar( + select(ScheduleRuns).where(ScheduleRuns.idempotency_key == idempotency_key) + ) + if existing is not None: + if ( + existing.workspace_id != workspace_id + or existing.schedule_id != schedule_id + ): + raise TriggerError("Idempotency-Key belongs to another schedule run") + return existing, False + + # 2) Lock + load schedule for the duration of this transaction. + schedule = await _load_schedule( + session, schedule_id, workspace_id, for_update=True, + ) + + # 3) Build snapshot (validates node arguments eagerly). + node_rows, edges = await _load_graph(session, schedule_id) + snapshot = _build_snapshot(schedule, node_rows, edges) + + # 4) Validate DAG after snapshot so parse_node_arguments errors + # surface first. + errors = _validate_dag( + [n for n, _v, _s in node_rows], edges, + ) + if errors: + raise InvalidDag(errors) + + # 5) Persist run + outbox. + now = utcnow() + run = ScheduleRuns( + run_id=new_ulid(), + schedule_id=schedule.schedule_id, + workspace_id=schedule.workspace_id, + workflow_version=schedule.workflow_version, + trigger_type=trigger_type, + idempotency_key=idempotency_key, + run_status="queued", + state_version=0, + schedule_snapshot=snapshot, + queued_at=now, + triggered_by=triggered_by_user_id, + ) + session.add(run) + schedule.last_run_at = now + + await add_outbox_event( + session, + event_type="schedule.run.requested", + producer="platform-api", + trace_id=trace_id, + aggregate_type="schedule_run", + aggregate_id=run.run_id, + idempotency_key=idempotency_key, + payload={ + "workspace_id": run.workspace_id, + "schedule_id": run.schedule_id, + "run_id": run.run_id, + "workflow_version": run.workflow_version, + "trigger_type": run.trigger_type, + "triggered_by": run.triggered_by, + "schedule_snapshot": snapshot, + }, + ) + await session.flush() + return run, True + + +__all__ = [ + "DagTooLarge", + "InvalidDag", + "InvalidNodeArguments", + "SYSTEM_CRON_USER_ID", + "ScheduleNotFound", + "TriggerError", + "create_scheduled_run", + "normalize_idempotency_key", + "parse_node_arguments", +]