feat(identifiers): add reversible projection migration
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
This commit is contained in:
parent
ce52e9d1e2
commit
cb1b028fd1
18 changed files with 921 additions and 27 deletions
308
api/services/work_record_identifier_migration.py
Normal file
308
api/services/work_record_identifier_migration.py
Normal file
|
|
@ -0,0 +1,308 @@
|
|||
"""Transactional State Hub projection migration for deterministic work-record UUIDs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select, text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from api.models.work_record_identifier_alias import WorkRecordIdentifierAlias
|
||||
|
||||
PLAN_SCHEMA = "repo-manager.identifier-migration-plan.v1"
|
||||
FLEET_NAMESPACE = "helixforge"
|
||||
DERIVATION_NAMESPACE_UUID = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009")
|
||||
_RECORD_ID_RE = re.compile(r"^[A-Z][A-Z0-9-]*-WP-[0-9]{4}(?:-T[0-9]{2,})?$")
|
||||
|
||||
|
||||
class IdentifierMigrationError(ValueError):
|
||||
"""The signed plan or current projection is unsafe to migrate."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class IdentifierMigrationResult:
|
||||
repo_slug: str
|
||||
plan_sha256: str
|
||||
direction: str
|
||||
replacements: int
|
||||
assignments_deferred: int
|
||||
|
||||
|
||||
def verify_plan(plan: dict[str, Any]) -> str:
|
||||
"""Validate the immutable plan envelope and deterministic UUID mappings."""
|
||||
if plan.get("schema") != PLAN_SCHEMA:
|
||||
raise IdentifierMigrationError("unsupported identifier migration plan schema")
|
||||
if plan.get("namespace") != FLEET_NAMESPACE:
|
||||
raise IdentifierMigrationError(
|
||||
f"plan namespace must be {FLEET_NAMESPACE!r}"
|
||||
)
|
||||
if plan.get("apply_policy") != "all-or-nothing per repository":
|
||||
raise IdentifierMigrationError("plan does not declare repository-atomic apply")
|
||||
if plan.get("ready_to_apply") is not True:
|
||||
raise IdentifierMigrationError("plan is not ready_to_apply")
|
||||
|
||||
expected = plan.get("plan_sha256")
|
||||
if not isinstance(expected, str) or not re.fullmatch(r"[0-9a-f]{64}", expected):
|
||||
raise IdentifierMigrationError("plan has no valid SHA-256 seal")
|
||||
unsealed = {key: value for key, value in plan.items() if key != "plan_sha256"}
|
||||
canonical = json.dumps(unsealed, sort_keys=True, separators=(",", ":")).encode()
|
||||
actual = hashlib.sha256(canonical).hexdigest()
|
||||
if actual != expected:
|
||||
raise IdentifierMigrationError("plan SHA-256 mismatch")
|
||||
return expected
|
||||
|
||||
|
||||
def _repository_mappings(
|
||||
plan: dict[str, Any], repo_slug: str
|
||||
) -> tuple[list[dict[str, Any]], int]:
|
||||
matches = [item for item in plan.get("repositories", []) if item.get("repo") == repo_slug]
|
||||
if len(matches) != 1:
|
||||
raise IdentifierMigrationError(
|
||||
f"repository {repo_slug!r} must occur exactly once in plan"
|
||||
)
|
||||
repository = matches[0]
|
||||
if repository.get("eligible") is not True or repository.get("atomic_unit") is not True:
|
||||
raise IdentifierMigrationError(f"repository {repo_slug!r} is not eligible and atomic")
|
||||
|
||||
replacements: list[dict[str, Any]] = []
|
||||
assignments = 0
|
||||
seen_old: set[uuid.UUID] = set()
|
||||
seen_new: set[uuid.UUID] = set()
|
||||
for raw in repository.get("mappings", []):
|
||||
action = raw.get("action")
|
||||
if action == "assign":
|
||||
assignments += 1
|
||||
continue
|
||||
if action == "unchanged":
|
||||
continue
|
||||
if action != "replace":
|
||||
raise IdentifierMigrationError(f"unsupported mapping action {action!r}")
|
||||
if raw.get("repo") != repo_slug:
|
||||
raise IdentifierMigrationError("mapping repository does not match atomic unit")
|
||||
kind = raw.get("kind")
|
||||
record_id = raw.get("record_id")
|
||||
if kind not in {"workplan", "task"}:
|
||||
raise IdentifierMigrationError(f"unsupported record kind {kind!r}")
|
||||
if not isinstance(record_id, str) or not _RECORD_ID_RE.fullmatch(record_id):
|
||||
raise IdentifierMigrationError(f"noncanonical record id {record_id!r}")
|
||||
try:
|
||||
old_id = uuid.UUID(str(raw.get("current_uuid")))
|
||||
new_id = uuid.UUID(str(raw.get("derived_uuid")))
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise IdentifierMigrationError(f"invalid UUID mapping for {record_id}") from exc
|
||||
expected_new = uuid.uuid5(
|
||||
DERIVATION_NAMESPACE_UUID,
|
||||
f"{FLEET_NAMESPACE}\n{record_id}",
|
||||
)
|
||||
if new_id != expected_new:
|
||||
raise IdentifierMigrationError(f"derived UUID mismatch for {record_id}")
|
||||
if old_id == new_id or old_id in seen_old or new_id in seen_new:
|
||||
raise IdentifierMigrationError(f"non-unique replacement for {record_id}")
|
||||
seen_old.add(old_id)
|
||||
seen_new.add(new_id)
|
||||
replacements.append(
|
||||
{
|
||||
"kind": kind,
|
||||
"record_id": record_id,
|
||||
"old_id": old_id,
|
||||
"new_id": new_id,
|
||||
}
|
||||
)
|
||||
return replacements, assignments
|
||||
|
||||
|
||||
async def _assert_projection_preconditions(
|
||||
session: AsyncSession,
|
||||
repo_slug: str,
|
||||
replacements: list[dict[str, Any]],
|
||||
*,
|
||||
reverse: bool,
|
||||
) -> None:
|
||||
for mapping in replacements:
|
||||
source_id = mapping["new_id"] if reverse else mapping["old_id"]
|
||||
target_id = mapping["old_id"] if reverse else mapping["new_id"]
|
||||
if mapping["kind"] == "workplan":
|
||||
source_query = text(
|
||||
"SELECT workplans.id FROM workplans "
|
||||
"JOIN managed_repos ON managed_repos.id = workplans.repo_id "
|
||||
"WHERE managed_repos.slug = :repo_slug AND workplans.id = :record_id "
|
||||
"FOR UPDATE"
|
||||
)
|
||||
target_query = text("SELECT id FROM workplans WHERE id = :record_id")
|
||||
else:
|
||||
source_query = text(
|
||||
"SELECT tasks.id FROM tasks "
|
||||
"JOIN workplans ON workplans.id = tasks.workplan_id "
|
||||
"JOIN managed_repos ON managed_repos.id = workplans.repo_id "
|
||||
"WHERE managed_repos.slug = :repo_slug AND tasks.id = :record_id "
|
||||
"FOR UPDATE"
|
||||
)
|
||||
target_query = text("SELECT id FROM tasks WHERE id = :record_id")
|
||||
source = await session.execute(
|
||||
source_query,
|
||||
{"repo_slug": repo_slug, "record_id": source_id},
|
||||
)
|
||||
if source.scalar_one_or_none() is None:
|
||||
raise IdentifierMigrationError(
|
||||
f"source {mapping['kind']} {source_id} is absent from repository {repo_slug}"
|
||||
)
|
||||
target = await session.execute(target_query, {"record_id": target_id})
|
||||
if target.scalar_one_or_none() is not None:
|
||||
raise IdentifierMigrationError(
|
||||
f"target {mapping['kind']} {target_id} already exists"
|
||||
)
|
||||
|
||||
|
||||
async def apply_repository_identifier_migration(
|
||||
session: AsyncSession,
|
||||
plan: dict[str, Any],
|
||||
repo_slug: str,
|
||||
) -> IdentifierMigrationResult:
|
||||
"""Replace one repository's projected UUIDs in one database transaction."""
|
||||
plan_sha256 = verify_plan(plan)
|
||||
replacements, assignments = _repository_mappings(plan, repo_slug)
|
||||
if session.in_transaction():
|
||||
raise IdentifierMigrationError("migration requires a fresh database session")
|
||||
|
||||
async with session.begin():
|
||||
await _assert_projection_preconditions(
|
||||
session, repo_slug, replacements, reverse=False
|
||||
)
|
||||
aliases = {
|
||||
alias.old_id: alias
|
||||
for alias in (
|
||||
await session.execute(
|
||||
select(WorkRecordIdentifierAlias).where(
|
||||
WorkRecordIdentifierAlias.old_id.in_(
|
||||
[mapping["old_id"] for mapping in replacements]
|
||||
)
|
||||
)
|
||||
)
|
||||
).scalars()
|
||||
}
|
||||
for mapping in replacements:
|
||||
alias = aliases.get(mapping["old_id"])
|
||||
if alias is None:
|
||||
alias = WorkRecordIdentifierAlias(
|
||||
old_id=mapping["old_id"],
|
||||
new_id=mapping["new_id"],
|
||||
record_kind=mapping["kind"],
|
||||
record_id=mapping["record_id"],
|
||||
repo_slug=repo_slug,
|
||||
namespace=FLEET_NAMESPACE,
|
||||
plan_sha256=plan_sha256,
|
||||
)
|
||||
session.add(alias)
|
||||
aliases[mapping["old_id"]] = alias
|
||||
elif (
|
||||
alias.new_id != mapping["new_id"]
|
||||
or alias.record_kind != mapping["kind"]
|
||||
or alias.record_id != mapping["record_id"]
|
||||
or alias.repo_slug != repo_slug
|
||||
or alias.namespace != FLEET_NAMESPACE
|
||||
or alias.plan_sha256 != plan_sha256
|
||||
or alias.migration_status != "reversed"
|
||||
):
|
||||
raise IdentifierMigrationError(
|
||||
f"conflicting durable alias for {mapping['old_id']}"
|
||||
)
|
||||
await session.flush()
|
||||
|
||||
for kind in ("workplan", "task"):
|
||||
table = "workplans" if kind == "workplan" else "tasks"
|
||||
for mapping in (item for item in replacements if item["kind"] == kind):
|
||||
result = await session.execute(
|
||||
text(f"UPDATE {table} SET id = :new_id WHERE id = :old_id"),
|
||||
{"old_id": mapping["old_id"], "new_id": mapping["new_id"]},
|
||||
)
|
||||
if result.rowcount != 1:
|
||||
raise IdentifierMigrationError(
|
||||
f"failed to replace {mapping['record_id']}"
|
||||
)
|
||||
|
||||
now = datetime.now(UTC)
|
||||
for alias in aliases.values():
|
||||
alias.migration_status = "applied"
|
||||
alias.applied_at = now
|
||||
alias.reversed_at = None
|
||||
|
||||
return IdentifierMigrationResult(
|
||||
repo_slug=repo_slug,
|
||||
plan_sha256=plan_sha256,
|
||||
direction="forward",
|
||||
replacements=len(replacements),
|
||||
assignments_deferred=assignments,
|
||||
)
|
||||
|
||||
|
||||
async def reverse_repository_identifier_migration(
|
||||
session: AsyncSession,
|
||||
plan: dict[str, Any],
|
||||
repo_slug: str,
|
||||
) -> IdentifierMigrationResult:
|
||||
"""Restore one repository's prior projected UUIDs and retain aliases."""
|
||||
plan_sha256 = verify_plan(plan)
|
||||
replacements, assignments = _repository_mappings(plan, repo_slug)
|
||||
if session.in_transaction():
|
||||
raise IdentifierMigrationError("migration requires a fresh database session")
|
||||
|
||||
async with session.begin():
|
||||
aliases = list(
|
||||
(
|
||||
await session.execute(
|
||||
select(WorkRecordIdentifierAlias).where(
|
||||
WorkRecordIdentifierAlias.plan_sha256 == plan_sha256,
|
||||
WorkRecordIdentifierAlias.repo_slug == repo_slug,
|
||||
)
|
||||
)
|
||||
).scalars()
|
||||
)
|
||||
aliases_by_old = {alias.old_id: alias for alias in aliases}
|
||||
if len(aliases_by_old) != len(replacements):
|
||||
raise IdentifierMigrationError("durable alias set is incomplete")
|
||||
for mapping in replacements:
|
||||
alias = aliases_by_old.get(mapping["old_id"])
|
||||
if (
|
||||
alias is None
|
||||
or alias.new_id != mapping["new_id"]
|
||||
or alias.record_kind != mapping["kind"]
|
||||
or alias.record_id != mapping["record_id"]
|
||||
or alias.migration_status != "applied"
|
||||
):
|
||||
raise IdentifierMigrationError(
|
||||
f"durable alias is not applied for {mapping['record_id']}"
|
||||
)
|
||||
await _assert_projection_preconditions(
|
||||
session, repo_slug, replacements, reverse=True
|
||||
)
|
||||
|
||||
for kind in ("task", "workplan"):
|
||||
table = "tasks" if kind == "task" else "workplans"
|
||||
for mapping in (item for item in replacements if item["kind"] == kind):
|
||||
result = await session.execute(
|
||||
text(f"UPDATE {table} SET id = :old_id WHERE id = :new_id"),
|
||||
{"old_id": mapping["old_id"], "new_id": mapping["new_id"]},
|
||||
)
|
||||
if result.rowcount != 1:
|
||||
raise IdentifierMigrationError(
|
||||
f"failed to reverse {mapping['record_id']}"
|
||||
)
|
||||
|
||||
now = datetime.now(UTC)
|
||||
for alias in aliases:
|
||||
alias.migration_status = "reversed"
|
||||
alias.reversed_at = now
|
||||
|
||||
return IdentifierMigrationResult(
|
||||
repo_slug=repo_slug,
|
||||
plan_sha256=plan_sha256,
|
||||
direction="reverse",
|
||||
replacements=len(replacements),
|
||||
assignments_deferred=assignments,
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue