dify/api/services/workflow_version_number_service.py

"""Allocation of user-facing workflow version numbers (`#N`).

Numbers are unique and monotonically increasing within an app, and are never
reused: `workflow_version_counters` keeps the highest number handed out so far,
so deleting a published version does not free its number.

The counter is keyed by `Workflow.app_id`, which is polymorphic — it holds an app
id, a pipeline id or a snippet id depending on the workflow kind. UUID uniqueness
across those tables is why the same counter table serves all of them.
"""

from sqlalchemy import select
from sqlalchemy.dialects.mysql import insert as mysql_insert
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

from configs import dify_config
from models.workflow import WorkflowVersionCounter


def allocate_version_number(*, session: Session, app_id: str) -> int:
    """Reserve and return the next version number for `app_id`.

    The upsert acquires a row lock that is held until the caller's transaction
    commits, so concurrent publishes of the same app serialize and never receive
    the same number. Callers must run inside a transaction; if it rolls back,
    both the counter update and workflow creation roll back, leaving the number
    available for the next successful publish.
    """
    # Dialect-specific upsert, mirroring `workflow_draft_variable_service`: the
    # ORM cannot express "insert or increment" and a read-then-write would race.
    # PostgreSQL returns the new value inline; MySQL has no RETURNING, so the
    # value is read back within the same transaction while the row is still
    # locked by the upsert.
    if dify_config.SQLALCHEMY_DATABASE_URI_SCHEME == "postgresql":
        stmt = (
            pg_insert(WorkflowVersionCounter)
            .values(app_id=app_id, last_version_number=1)
            .on_conflict_do_update(
                index_elements=[WorkflowVersionCounter.app_id],
                set_={"last_version_number": WorkflowVersionCounter.last_version_number + 1},
            )
            .returning(WorkflowVersionCounter.last_version_number)
        )
        version_number = session.scalar(stmt)
    else:
        insert_stmt = mysql_insert(WorkflowVersionCounter).values(app_id=app_id, last_version_number=1)
        session.execute(
            insert_stmt.on_duplicate_key_update(  # type: ignore[attr-defined]
                last_version_number=WorkflowVersionCounter.last_version_number + 1,
            )
        )
        version_number = session.scalar(
            select(WorkflowVersionCounter.last_version_number).where(WorkflowVersionCounter.app_id == app_id)
        )

    if version_number is None:
        raise ValueError(f"Failed to allocate a workflow version number for app {app_id}.")
    return version_number