"""
Service for the persistent strategy model.
Implements the versioning rules for the persistent strategy model:
* Creating uses ``version=1`` and ``status=DRAFT``.
* Editing a draft mutates the row in place.
* Editing a published row forks ``version + 1`` as a new draft instead
of mutating the published copy.
* Publishing a draft transitions it to ``PUBLISHED`` and archives any
sibling that was previously published, so a ``(realmId, name)``
family only ever has at most one live row.
* Archiving moves a row out of the active set without deleting history.
Tenancy is enforced at this layer: every read/write takes ``realmId``
and we never accept it from the caller body - the endpoint resolves it
from the auth context and passes it in.
"""
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import List, Optional
from app.core.exceptions import (
BadRequestError,
ConflictError,
DuplicatedError,
NotFoundError,
)
from app.engine.dsl_validator import validate_ast
from app.model.strategy_definition import (
StrategyDefinition,
StrategyDefinitionStatus,
StrategyDefinitionType,
)
from app.repository.strategy_definition_repository import StrategyDefinitionRepository
from app.schema.strategy_definition_schema import (
StrategyDefinitionCreate,
StrategyDefinitionPersist,
StrategyDefinitionRead,
StrategyDefinitionUpdate,
StrategyUsageGame,
StrategyUsageRead,
StrategyUsageTask,
)
from app.services.base_service import BaseService
# Kept in sync with ``CUSTOM_STRATEGY_PREFIX`` in
# ``app/services/strategy_service.py``. We inline the literal here
# instead of importing it because ``StrategyService`` already imports
# this module - pulling the constant back the other way would create a
# circular module dependency (same reason ``_validate_payload`` doesn't
# resolve parent strategies; see comment below).
_CUSTOM_STRATEGY_PREFIX = "custom:"
[docs]
@dataclass(frozen=True)
class RollbackResult:
"""Outcome of a rollback operation, returned by the service so the
endpoint can include cascade counts in the audit log without re-hitting
the DB."""
strategy: StrategyDefinitionRead
games_reassigned: int
tasks_reassigned: int
[docs]
class StrategyDefinitionService(BaseService):
"""
CRUD + lifecycle operations for custom strategies.
"""
[docs]
def __init__(
self,
strategy_definition_repository: StrategyDefinitionRepository,
game_repository=None,
task_repository=None,
) -> None:
"""
``game_repository`` and ``task_repository`` are optional so legacy
call sites that only need CRUD/lifecycle (most tests, the
simulation service) keep working without a wider DI graph. The
The rollback flow requires both - when missing,
:meth:`rollback` raises a precise error rather than silently
leaving cascade UPDATEs undone.
"""
self.strategy_definition_repository = strategy_definition_repository
self.game_repository = game_repository
self.task_repository = task_repository
super().__init__(strategy_definition_repository)
@staticmethod
def _to_read(row: StrategyDefinition) -> StrategyDefinitionRead:
"""
Map a ``StrategyDefinition`` ORM row to its public read schema.
Args:
row (StrategyDefinition): The persisted definition row.
Returns:
StrategyDefinitionRead: The API-facing representation.
"""
return StrategyDefinitionRead(
id=str(row.id),
realmId=row.realmId,
name=row.name,
description=row.description,
type=row.type,
parentStrategyId=row.parentStrategyId,
astJson=row.astJson,
blocklyXml=row.blocklyXml,
version=row.version,
status=row.status,
createdBy=row.createdBy,
created_at=row.created_at,
updated_at=row.updated_at,
publishedAt=row.publishedAt,
experimentTag=row.experimentTag,
)
@staticmethod
def _validate_payload(type_value: str, parentStrategyId: Optional[str]) -> None:
"""
DSL_EXTEND must carry a parent; DSL_FULL must not. We don't yet
validate that ``parentStrategyId`` resolves to a real registry
entry because the registry is loaded lazily and importing it
here would create a circular dependency - that check lives in
the endpoint via :class:`StrategyService`.
"""
if type_value == StrategyDefinitionType.DSL_EXTEND.value:
if not parentStrategyId:
raise BadRequestError(
detail="DSL_EXTEND strategies require parentStrategyId."
)
else:
if parentStrategyId:
raise BadRequestError(
detail=(
f"parentStrategyId is only valid when type="
f"{StrategyDefinitionType.DSL_EXTEND.value}."
)
)
[docs]
async def name_exists(
self,
*,
realmId: Optional[str],
name: str,
) -> bool:
"""Whether any version of ``(realmId, name)`` already exists.
Used by the import endpoint to decide whether the
incoming bundle needs an auto-rename to avoid colliding with
the ``UNIQUE(realmId, name, version)`` constraint.
"""
max_version = await self.strategy_definition_repository.get_max_version(
realmId=realmId, name=name
)
return max_version > 0
[docs]
async def list_strategies(
self,
*,
realmId: Optional[str],
status: Optional[str] = None,
type: Optional[str] = None,
limit: int = 100,
) -> List[StrategyDefinitionRead]:
"""
List strategy definitions for a realm, optionally filtered.
Args:
realmId (Optional[str]): Realm/tenant to scope to.
status (Optional[str]): Optional lifecycle-status filter.
type (Optional[str]): Optional strategy-type filter.
limit (int): Maximum rows to return.
Returns:
List[StrategyDefinitionRead]: The matching definitions.
"""
rows = await self.strategy_definition_repository.list_for_realm(
realmId=realmId, status=status, type=type, limit=limit
)
return [self._to_read(r) for r in rows]
[docs]
async def get_strategy(
self,
*,
id: str,
realmId: Optional[str],
) -> StrategyDefinitionRead:
"""
Fetch one strategy definition scoped to a realm.
Args:
id (str): Strategy definition identifier.
realmId (Optional[str]): Realm/tenant the strategy must belong to.
Returns:
StrategyDefinitionRead: The matching definition.
Raises:
NotFoundError: If no matching strategy exists in the realm.
"""
row = await self.strategy_definition_repository.get_for_realm(
id=id, realmId=realmId
)
if row is None:
raise NotFoundError(detail=f"Custom strategy not found: {id}")
return self._to_read(row)
[docs]
async def create(
self,
*,
payload: StrategyDefinitionCreate,
realmId: Optional[str],
createdBy: Optional[str],
apiKey_used: Optional[str],
oauth_user_id: Optional[str],
) -> StrategyDefinitionRead:
"""
Create a new strategy-definition draft.
Validates the type/parent combination and any provided AST, rejects a
name that already exists in the realm, and persists the draft as
version 1.
Args:
payload (StrategyDefinitionCreate): The strategy to create.
realmId (Optional[str]): Realm/tenant that will own it.
createdBy (Optional[str]): Identity recorded as the author.
apiKey_used (Optional[str]): API key used for the request, if any.
oauth_user_id (Optional[str]): OAuth subject, if any.
Returns:
StrategyDefinitionRead: The newly created draft.
Raises:
BadRequestError: If the type/parent combination is invalid.
DuplicatedError: If a strategy with the same name already exists.
"""
type_value = payload.type.value
self._validate_payload(type_value, payload.parentStrategyId)
if payload.astJson is not None:
validate_ast(payload.astJson)
existing_max = await self.strategy_definition_repository.get_max_version(
realmId=realmId, name=payload.name
)
if existing_max > 0:
raise DuplicatedError(
detail=(
f"A strategy named '{payload.name}' already exists in "
"this realm. Edit the existing one to create a new "
"version."
)
)
persist = StrategyDefinitionPersist(
realmId=realmId,
name=payload.name,
description=payload.description,
type=type_value,
parentStrategyId=payload.parentStrategyId,
astJson=payload.astJson,
blocklyXml=payload.blocklyXml,
version=1,
status=StrategyDefinitionStatus.DRAFT.value,
createdBy=createdBy,
experimentTag=payload.experimentTag,
apiKey_used=apiKey_used,
oauth_user_id=oauth_user_id,
)
row = await self.strategy_definition_repository.create(persist)
return self._to_read(row)
[docs]
async def update(
self,
*,
id: str,
payload: StrategyDefinitionUpdate,
realmId: Optional[str],
createdBy: Optional[str],
apiKey_used: Optional[str],
oauth_user_id: Optional[str],
) -> StrategyDefinitionRead:
"""
Apply an update.
* On a DRAFT row: patch in place and return the same id.
* On a PUBLISHED row: fork a new DRAFT at ``version + 1`` with
the patched fields applied, leaving the published row
untouched so it keeps running until an explicit publish.
* On an ARCHIVED row: refuse - archived strategies are
immutable.
"""
row = await self.strategy_definition_repository.get_for_realm(
id=id, realmId=realmId
)
if row is None:
raise NotFoundError(detail=f"Custom strategy not found: {id}")
if row.status == StrategyDefinitionStatus.ARCHIVED.value:
raise ConflictError(
detail=(
"Archived strategies cannot be edited. Clone the "
"definition into a new strategy if you need to "
"iterate on it."
)
)
merged = {
"name": payload.name if payload.name is not None else row.name,
"description": (
payload.description
if payload.description is not None
else row.description
),
"type": (payload.type.value if payload.type is not None else row.type),
"parentStrategyId": (
payload.parentStrategyId
if payload.parentStrategyId is not None
else row.parentStrategyId
),
"astJson": (
payload.astJson if payload.astJson is not None else row.astJson
),
"blocklyXml": (
payload.blocklyXml if payload.blocklyXml is not None else row.blocklyXml
),
"experimentTag": (
payload.experimentTag
if payload.experimentTag is not None
else row.experimentTag
),
}
self._validate_payload(merged["type"], merged["parentStrategyId"])
if merged["astJson"] is not None:
validate_ast(merged["astJson"])
if row.status == StrategyDefinitionStatus.DRAFT.value:
patch = StrategyDefinitionUpdate(
name=merged["name"],
description=merged["description"],
type=merged["type"],
parentStrategyId=merged["parentStrategyId"],
astJson=merged["astJson"],
blocklyXml=merged["blocklyXml"],
experimentTag=merged["experimentTag"],
)
updated = await self.strategy_definition_repository.update(id, patch)
return self._to_read(updated)
# PUBLISHED → fork a new draft with the merged contents.
next_version = (
await self.strategy_definition_repository.get_max_version(
realmId=realmId, name=row.name
)
) + 1
persist = StrategyDefinitionPersist(
realmId=realmId,
name=merged["name"],
description=merged["description"],
type=merged["type"],
parentStrategyId=merged["parentStrategyId"],
astJson=merged["astJson"],
blocklyXml=merged["blocklyXml"],
version=next_version,
status=StrategyDefinitionStatus.DRAFT.value,
createdBy=createdBy,
experimentTag=merged["experimentTag"],
apiKey_used=apiKey_used,
oauth_user_id=oauth_user_id,
)
new_row = await self.strategy_definition_repository.create(persist)
return self._to_read(new_row)
[docs]
async def publish(
self,
*,
id: str,
realmId: Optional[str],
) -> StrategyDefinitionRead:
"""
Publish a draft strategy, archiving any previously-published sibling.
Idempotent: re-publishing an already-published row is a no-op. If
another version of the same name is published, it is archived first so
only one published version exists per name.
Args:
id (str): Strategy definition identifier.
realmId (Optional[str]): Realm/tenant the strategy must belong to.
Returns:
StrategyDefinitionRead: The published strategy.
Raises:
NotFoundError: If no matching strategy exists in the realm.
ConflictError: If the strategy is archived.
"""
row = await self.strategy_definition_repository.get_for_realm(
id=id, realmId=realmId
)
if row is None:
raise NotFoundError(detail=f"Custom strategy not found: {id}")
if row.status == StrategyDefinitionStatus.ARCHIVED.value:
raise ConflictError(detail="Cannot publish an archived strategy.")
if row.status == StrategyDefinitionStatus.PUBLISHED.value:
# Idempotent: re-publishing the same row is a no-op.
return self._to_read(row)
sibling = await self.strategy_definition_repository.get_published(
realmId=realmId, name=row.name
)
if sibling is not None and str(sibling.id) != str(row.id):
await self.strategy_definition_repository.set_status(
id=str(sibling.id),
status=StrategyDefinitionStatus.ARCHIVED.value,
)
await self.strategy_definition_repository.set_status(
id=str(row.id),
status=StrategyDefinitionStatus.PUBLISHED.value,
publishedAt=datetime.now(timezone.utc),
)
return await self.get_strategy(id=id, realmId=realmId)
[docs]
async def archive(
self,
*,
id: str,
realmId: Optional[str],
) -> StrategyDefinitionRead:
"""
Archive a strategy so it can no longer be published or assigned.
Idempotent: archiving an already-archived row returns it unchanged.
Args:
id (str): Strategy definition identifier.
realmId (Optional[str]): Realm/tenant the strategy must belong to.
Returns:
StrategyDefinitionRead: The archived strategy.
Raises:
NotFoundError: If no matching strategy exists in the realm.
"""
row = await self.strategy_definition_repository.get_for_realm(
id=id, realmId=realmId
)
if row is None:
raise NotFoundError(detail=f"Custom strategy not found: {id}")
if row.status == StrategyDefinitionStatus.ARCHIVED.value:
return self._to_read(row)
await self.strategy_definition_repository.set_status(
id=str(row.id),
status=StrategyDefinitionStatus.ARCHIVED.value,
)
return await self.get_strategy(id=id, realmId=realmId)
[docs]
async def list_versions(
self,
*,
id: str,
realmId: Optional[str],
) -> List[StrategyDefinitionRead]:
"""
Return every version in the family that contains ``id``, newest
first. The caller passes a single id (typically the current
published version) and we resolve the family name from it so the
endpoint contract stays "one id in, full history out".
Tenant-scoped: ``get_for_realm`` 404s when the row belongs to
another realm, so cross-tenant probing returns nothing.
"""
row = await self.strategy_definition_repository.get_for_realm(
id=id, realmId=realmId
)
if row is None:
raise NotFoundError(detail=f"Custom strategy not found: {id}")
rows = await self.strategy_definition_repository.list_versions(
realmId=realmId, name=row.name
)
return [self._to_read(r) for r in rows]
[docs]
async def get_usage(
self,
*,
id: str,
realmId: Optional[str],
) -> StrategyUsageRead:
"""
Reverse lookup: which games/tasks are assigned to this exact
strategy version.
Consumers store the assignable id ``custom:<uuid>``, so usage is
an exact match on that string - the same value the rollback
cascade rewrites. We report per-version (not per-family) because
each published version is a distinct uuid that games point at
individually; that matches what an admin needs to see before
reassigning, archiving or rolling back *this* version.
Tenant-scoped via ``get_for_realm``: a cross-realm id 404s.
No cross-tenant leak through the usage lists either - a game can
only be assigned a strategy validated to live in its own realm,
so every consumer of ``custom:<uuid>`` shares the strategy's
realm.
"""
if self.game_repository is None or self.task_repository is None:
raise BadRequestError(
detail=(
"Usage lookup requires game/task repositories wired "
"into StrategyDefinitionService."
)
)
row = await self.strategy_definition_repository.get_for_realm(
id=id, realmId=realmId
)
if row is None:
raise NotFoundError(detail=f"Custom strategy not found: {id}")
assignable_id = f"{_CUSTOM_STRATEGY_PREFIX}{row.id}"
game_rows = await self.game_repository.list_by_strategy_id(assignable_id)
task_rows = await self.task_repository.list_by_strategy_id(assignable_id)
# Resolve each task's parent game external id in one batched query
# so the UI can show "task X of game Y" rather than a raw UUID.
parent_ids = {t.gameId for t in task_rows if t.gameId is not None}
external_by_game = await self.game_repository.list_external_ids(parent_ids)
games = [
StrategyUsageGame(
gameId=str(g.id),
externalGameId=g.externalGameId,
platform=g.platform,
)
for g in game_rows
]
tasks = [
StrategyUsageTask(
taskId=str(t.id),
externalTaskId=t.externalTaskId,
gameId=str(t.gameId) if t.gameId is not None else None,
externalGameId=external_by_game.get(t.gameId),
)
for t in task_rows
]
return StrategyUsageRead(
strategyId=assignable_id,
name=row.name,
version=row.version,
status=row.status,
gameCount=len(games),
taskCount=len(tasks),
games=games,
tasks=tasks,
)
[docs]
async def rollback(
self,
*,
id: str,
target_version: int,
realmId: Optional[str],
) -> RollbackResult:
"""
Promote ``target_version`` back to PUBLISHED in the family that
contains ``id``, archive whichever version is currently published,
and rewrite every ``Games.strategyId`` / ``Tasks.strategyId``
pointing at the displaced row so no consumer is left referencing
an ARCHIVED row.
``id`` doesn't have to be PUBLISHED: an admin may initiate rollback
from any version of the family (e.g. browsing the version history
UI). We always treat the family's *current* PUBLISHED row as the
one to archive + reassign, regardless of which id was clicked.
Errors:
* 404 if ``id`` doesn't resolve (or is in another realm).
* 404 if ``target_version`` isn't a version of that family.
* 409 if the requested target is the row that's already PUBLISHED
- rolling back to the current state would be a no-op that
falsely advertises a status change in the audit log.
"""
if self.game_repository is None or self.task_repository is None:
# Defensive: rollback writes to Games/Tasks. If those weren't
# wired we'd silently leave the cascade undone - much worse
# than refusing the operation up front.
raise BadRequestError(
detail=(
"Rollback requires game/task repositories wired into "
"StrategyDefinitionService."
)
)
current = await self.strategy_definition_repository.get_for_realm(
id=id, realmId=realmId
)
if current is None:
raise NotFoundError(detail=f"Custom strategy not found: {id}")
target = await self.strategy_definition_repository.get_version(
realmId=realmId,
name=current.name,
version=target_version,
)
if target is None:
raise NotFoundError(
detail=(
f"Version {target_version} not found for strategy "
f"'{current.name}'."
)
)
displaced = await self.strategy_definition_repository.get_published(
realmId=realmId, name=current.name
)
if displaced is not None and str(displaced.id) == str(target.id):
# Target is already the live PUBLISHED row - rollback would be
# a no-op and would obscure intent in the audit trail.
raise ConflictError(
detail=(
f"Version {target_version} is already the published "
f"version of '{current.name}'."
)
)
# Order matters: archive the displaced row first, then promote
# the target, then run the cascade. If a failure interrupts the
# sequence the worst observable state is "0 PUBLISHED in family"
# for a few ms - preferable to leaving two PUBLISHED rows.
if displaced is not None:
await self.strategy_definition_repository.set_status(
id=str(displaced.id),
status=StrategyDefinitionStatus.ARCHIVED.value,
)
await self.strategy_definition_repository.set_status(
id=str(target.id),
status=StrategyDefinitionStatus.PUBLISHED.value,
publishedAt=datetime.now(timezone.utc),
)
games_reassigned = 0
tasks_reassigned = 0
if displaced is not None:
old_strategy_id = f"{_CUSTOM_STRATEGY_PREFIX}{displaced.id}"
new_strategy_id = f"{_CUSTOM_STRATEGY_PREFIX}{target.id}"
games_reassigned = await self.game_repository.bulk_update_strategy_id(
old_strategy_id=old_strategy_id,
new_strategy_id=new_strategy_id,
)
tasks_reassigned = await self.task_repository.bulk_update_strategy_id(
old_strategy_id=old_strategy_id,
new_strategy_id=new_strategy_id,
)
promoted = await self.get_strategy(id=str(target.id), realmId=realmId)
return RollbackResult(
strategy=promoted,
games_reassigned=games_reassigned,
tasks_reassigned=tasks_reassigned,
)