from __future__ import annotations import base64 import hashlib import json import re from collections.abc import Callable from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from packages.pet_package_schema import ALLOWED_ACTIONS from services.ai.provider_metrics import sanitize_provider_call_metrics, summarize_provider_call_metrics from services.api.local_store import LocalJsonStore from services.api.state_guards import ( StateTransitionError, assert_can_approve_build, assert_can_create_build, assert_can_download_package, assert_can_mark_package_ready, assert_can_reserve_model_cost, assert_can_run_package, assert_can_start_full_build, assert_can_start_revision, ) from services.queue.local_build_queue import enqueue_build, find_next_queued_item BuildRunner = Callable[..., dict[str, Any]] MAX_UPLOAD_BYTES = 15 * 1024 * 1024 DEFAULT_PLAN_CODE = "sixxie_monthly_launch" MONTHLY_ENTITLEMENT_TYPES = ( "custom_pet_build", "revision_credit", "package_download", "package_runtime_access", ) class LocalPetApi: def __init__( self, *, store: LocalJsonStore, build_runner: BuildRunner, output_root: str | Path | None = None, media_storage: Any | None = None, ) -> None: self.store = store self.build_runner = build_runner self.output_root = Path(output_root) if output_root else store.path.parent / "pet_builds" self.media_storage = media_storage self.output_root.mkdir(parents=True, exist_ok=True) def health(self) -> dict[str, Any]: storage_health = self.media_storage.health() if self.media_storage is not None else {"ok": False} return { "ok": self.store.path.exists() and self.output_root.exists() and bool(storage_health.get("ok")), "store": "ok" if self.store.path.exists() else "missing", "storage": "ok" if storage_health.get("ok") else "missing", "output": "ok" if self.output_root.exists() else "missing", } def grant_test_subscription( self, *, user_id: str, pet_id: str | None = None, plan_code: str = DEFAULT_PLAN_CODE, source: str = "checkout", period_start: str | None = None, period_end: str | None = None, ) -> dict[str, Any]: state = self.store.transaction() now = _now() _ensure_user(state, user_id, now) if pet_id is not None: _require_owned_pet(state, user_id=user_id, pet_id=pet_id) _ensure_default_price_plan(state, now=now, plan_code=plan_code) starts_at = period_start or now expires_at = period_end or _one_month_later(now) subscription_id = _next_id(state, "subscription", "sub") provider = "invite" if source == "invite" else "stripe" state.setdefault("subscriptions", {})[subscription_id] = { "subscription_id": subscription_id, "user_id": user_id, "pet_id": pet_id, "plan_code": plan_code, "provider": provider, "source": source, "status": "active", "current_period_start": starts_at, "current_period_end": expires_at, "access_ends_at": expires_at, "cancel_at_period_end": False, "created_at": now, "updated_at": now, } order_id = None if source != "invite": order_id = _create_paid_order( state, user_id=user_id, pet_id=pet_id, subscription_id=subscription_id, provider="stripe", product_code=plan_code, amount_total_cents=999, now=now, ) entitlement_ids = _grant_monthly_entitlements( state, user_id=user_id, pet_id=pet_id, subscription_id=subscription_id, order_id=order_id, starts_at=starts_at, expires_at=expires_at, now=now, ) _append_event(state, user_id=user_id, pet_id=pet_id, event_name="entitlement_granted", now=now) self.store.write(state) return { "subscription_id": subscription_id, "entitlement_ids": entitlement_ids, "entitlement_types": list(MONTHLY_ENTITLEMENT_TYPES), "source": source, "user_visible_status": "subscription_active", "current_period_end": expires_at, } def create_test_invite( self, *, code: str, max_uses: int = 1, grant_plan_code: str = DEFAULT_PLAN_CODE, expires_at: str | None = None, ) -> dict[str, Any]: if not code.strip(): raise ValueError("invite code is required") if max_uses < 1: raise ValueError("max_uses must be at least 1") state = self.store.transaction() now = _now() invite_id = _next_id(state, "invite", "inv") state.setdefault("beta_invites", {})[invite_id] = { "invite_id": invite_id, "code_hash": _hash_invite_code(code), "status": "active", "max_uses": max_uses, "used_count": 0, "grant_plan_code": grant_plan_code, "grant_months": 1, "created_at": now, "expires_at": expires_at or _one_month_later(now), } self.store.write(state) return { "invite_id": invite_id, "user_visible_status": "invite_created", "max_uses": max_uses, } def redeem_test_invite(self, *, user_id: str, pet_id: str | None, code: str) -> dict[str, Any]: state = self.store.transaction() now = _now() _ensure_user(state, user_id, now) if pet_id is not None: _require_owned_pet(state, user_id=user_id, pet_id=pet_id) invite = _find_invite_by_code(state, code=code, now=now) redemption_key = f"{invite['invite_id']}:{user_id}" if redemption_key in state.setdefault("invite_redemptions", {}): raise StateTransitionError("invite has already been redeemed by this user") if int(invite["used_count"]) >= int(invite["max_uses"]): invite["status"] = "used_up" raise StateTransitionError("invite has no remaining uses") invite["used_count"] = int(invite["used_count"]) + 1 if int(invite["used_count"]) >= int(invite["max_uses"]): invite["status"] = "used_up" state["invite_redemptions"][redemption_key] = { "invite_id": invite["invite_id"], "user_id": user_id, "redeemed_at": now, } self.store.write(state) return self.grant_test_subscription( user_id=user_id, pet_id=pet_id, plan_code=str(invite.get("grant_plan_code") or DEFAULT_PLAN_CODE), source="invite", period_start=now, period_end=_one_month_later(now), ) def reserve_model_cost(self, *, user_id: str, pet_id: str, estimated_cost_cents: int) -> dict[str, Any]: state = self.store.transaction() _require_owned_pet(state, user_id=user_id, pet_id=pet_id) now = _now() quota = _get_or_create_model_cost_quota(state, user_id=user_id, pet_id=pet_id, now=now) current_cost = int(quota.get("used_value", 0)) + int(quota.get("reserved_value", 0)) assert_can_reserve_model_cost( current_cost_cents=current_cost, estimated_cost_cents=estimated_cost_cents, cap_cents=int(quota.get("limit_value", 500)), ) quota["reserved_value"] = int(quota.get("reserved_value", 0)) + estimated_cost_cents quota["updated_at"] = now usage_id = _next_id(state, "usage", "use") state.setdefault("usage_ledger", []).append( { "usage_id": usage_id, "quota_id": quota["quota_id"], "user_id": user_id, "pet_id": pet_id, "usage_type": "reserve", "amount": estimated_cost_cents, "balance_after": int(quota["reserved_value"]) + int(quota.get("used_value", 0)), "reason": "model_call_estimate", "created_at": now, } ) self.store.write(state) return { "quota_id": quota["quota_id"], "reserved_cost_cents": quota["reserved_value"], "cap_cents": quota["limit_value"], "user_visible_status": "model_cost_reserved", } def create_pet( self, *, user_id: str, pet_name: str, species: str | None = None, breed: str | None = None, personality_notes: str | None = None, typical_gestures: str | None = None, memory_notes: str | None = None, ) -> dict[str, Any]: if not pet_name.strip(): raise ValueError("pet_name is required") state = self.store.transaction() now = _now() _ensure_user(state, user_id, now) pet_id = _next_id(state, "pet", "pet") state["pet_profiles"][pet_id] = { "pet_id": pet_id, "user_id": user_id, "pet_name": pet_name.strip(), "species": species, "breed": breed, "personality_notes": personality_notes, "typical_gestures": typical_gestures, "memory_notes": memory_notes, "consent_status": "not_requested", "boundary_acknowledged": False, "status": "ready_for_media", "created_at": now, "updated_at": now, } _append_event(state, user_id=user_id, pet_id=pet_id, event_name="pet_profile_created", now=now) self.store.write(state) return { "pet_id": pet_id, "display_name": pet_name.strip(), "user_visible_status": "ready_for_media", "created_at": now, } def create_media_record( self, *, user_id: str, pet_id: str, media_type: str, mime_type: str, storage_ref: str, original_filename: str | None = None, ) -> dict[str, Any]: state = self.store.transaction() pet = _require_owned_pet(state, user_id=user_id, pet_id=pet_id) if media_type != "photo": raise ValueError("P0 local API only accepts photo media") now = _now() media_id = _next_id(state, "media", "med") state["pet_media"][media_id] = { "media_id": media_id, "pet_id": pet["pet_id"], "user_id": user_id, "media_type": media_type, "mime_type": mime_type, "original_filename": original_filename, "storage_ref": storage_ref, "source_quality": "unchecked", "deletion_status": "active", "created_at": now, "updated_at": now, } _append_event(state, user_id=user_id, pet_id=pet_id, event_name="pet_media_uploaded", now=now) self.store.write(state) return { "media_id": media_id, "user_visible_status": "uploaded", "quality_status": "unchecked", "created_at": now, } def upload_media_bytes( self, *, user_id: str, pet_id: str, filename: str, content_type: str, content: bytes, ) -> dict[str, Any]: if self.media_storage is None: raise ValueError("media storage is not configured") _assert_supported_image_upload(filename=filename, content_type=content_type, content=content) state = self.store.transaction() pet = _require_owned_pet(state, user_id=user_id, pet_id=pet_id) now = _now() media_id = _next_id(state, "media", "med") storage_ref = self.media_storage.put_media( user_id=user_id, pet_id=pet["pet_id"], media_id=media_id, filename=filename, content_type=content_type, content=content, ) state["pet_media"][media_id] = { "media_id": media_id, "pet_id": pet["pet_id"], "user_id": user_id, "media_type": "photo", "mime_type": content_type, "original_filename": filename, "storage_ref": storage_ref, "source_quality": "passed", "deletion_status": "active", "created_at": now, "updated_at": now, } _append_event(state, user_id=user_id, pet_id=pet_id, event_name="pet_media_uploaded", now=now) _append_event(state, user_id=user_id, pet_id=pet_id, event_name="pet_media_quality_passed", now=now) self.store.write(state) return { "media_id": media_id, "user_visible_status": "uploaded", "quality_status": "passed", "created_at": now, } def mark_media_quality(self, media_id: str, *, source_quality: str) -> dict[str, Any]: if source_quality not in {"passed", "failed"}: raise ValueError("source_quality must be passed or failed") state = self.store.transaction() media = _require(state["pet_media"], media_id, "media") now = _now() media["source_quality"] = source_quality media["updated_at"] = now _append_event( state, user_id=media["user_id"], pet_id=media["pet_id"], event_name="pet_media_quality_passed" if source_quality == "passed" else "pet_media_quality_failed", now=now, ) self.store.write(state) return { "media_id": media_id, "quality_status": source_quality, "updated_at": now, } def grant_consent( self, *, user_id: str, pet_id: str, boundary_acknowledged: bool, consent_version: str, external_processor_name: str | None = None, external_processor_consent: bool = False, external_processor_consent_version: str | None = None, ) -> dict[str, Any]: if boundary_acknowledged is not True: raise ValueError("boundary acknowledgement is required") state = self.store.transaction() pet = _require_owned_pet(state, user_id=user_id, pet_id=pet_id) now = _now() pet.update( { "consent_status": "granted", "consent_version": consent_version, "consent_at": now, "boundary_acknowledged": True, "external_processor_name": external_processor_name, "external_processor_consent": bool(external_processor_consent), "external_processor_consent_version": external_processor_consent_version, "external_processor_consent_at": now if external_processor_consent else None, "status": "generation_allowed", "updated_at": now, } ) _append_event(state, user_id=user_id, pet_id=pet_id, event_name="boundary_acknowledged", now=now) self.store.write(state) return { "pet_id": pet_id, "consent_status": "granted", "user_visible_status": "ready_to_generate", "updated_at": now, } def create_build( self, *, user_id: str, pet_id: str, media_ids: list[str], image_provider: str, image_model: str, ) -> dict[str, Any]: state = self.store.transaction() pet = _require_owned_pet(state, user_id=user_id, pet_id=pet_id) media = [_require_owned_media(state, user_id=user_id, pet_id=pet_id, media_id=media_id) for media_id in media_ids] if any(item.get("deletion_status") != "active" for item in media): raise StateTransitionError("active media is required before generation") assert_can_create_build( consent_status=pet["consent_status"], boundary_acknowledged=bool(pet.get("boundary_acknowledged")), media_quality_statuses=[item["source_quality"] for item in media], ) now = _now() build_entitlement = _find_active_entitlement( state, user_id=user_id, pet_id=pet_id, entitlement_type="custom_pet_build", now=now, ) assert_can_start_full_build( entitlement_status=str(build_entitlement.get("status") or "inactive"), quantity_total=int(build_entitlement.get("quantity_total", 0)), quantity_used=int(build_entitlement.get("quantity_used", 0)), expires_at=build_entitlement.get("expires_at"), now=now, ) _consume_entitlement(state, build_entitlement, now=now, event_user_id=user_id, event_pet_id=pet_id) build_id = _next_id(state, "build", "bld") state["pet_builds"][build_id] = { "build_id": build_id, "pet_id": pet_id, "user_id": user_id, "status": "queued", "input_media_ids": list(media_ids), "image_provider": image_provider, "image_model": image_model, "user_approved": False, "approval_answers": {}, "created_at": now, "updated_at": now, } enqueue_build(state, build_id=build_id, now=now) _append_event(state, user_id=user_id, pet_id=pet_id, build_id=build_id, event_name="pet_build_created", now=now) self.store.write(state) return { "build_id": build_id, "user_visible_status": "queued", "created_at": now, } def request_revision(self, build_id: str, *, user_id: str, revision_reason: str) -> dict[str, Any]: state = self.store.transaction() build = _require_owned_build(state, user_id=user_id, build_id=build_id) if build["status"] not in {"approved", "package_ready"} or build.get("user_approved") is not True: raise StateTransitionError("approved build with user approval is required before revision") now = _now() entitlement = _find_active_entitlement( state, user_id=user_id, pet_id=build["pet_id"], entitlement_type="revision_credit", now=now, ) assert_can_start_revision( entitlement_status=str(entitlement.get("status") or "inactive"), quantity_total=int(entitlement.get("quantity_total", 0)), quantity_used=int(entitlement.get("quantity_used", 0)), expires_at=entitlement.get("expires_at"), now=now, ) _consume_entitlement(state, entitlement, now=now, event_user_id=user_id, event_pet_id=build["pet_id"]) revision_build_id = _next_id(state, "build", "bld") state["pet_builds"][revision_build_id] = { "build_id": revision_build_id, "parent_build_id": build_id, "pet_id": build["pet_id"], "user_id": user_id, "status": "queued", "input_media_ids": list(build.get("input_media_ids") or []), "image_provider": build.get("image_provider"), "image_model": build.get("image_model"), "revision_reason": revision_reason[:220], "user_approved": False, "approval_answers": {}, "created_at": now, "updated_at": now, } enqueue_build(state, build_id=revision_build_id, now=now) _append_event( state, user_id=user_id, pet_id=build["pet_id"], build_id=revision_build_id, event_name="pet_build_created", now=now, ) self.store.write(state) return { "build_id": revision_build_id, "parent_build_id": build_id, "user_visible_status": "revision_queued", "created_at": now, } def process_build(self, build_id: str) -> dict[str, Any]: state = self.store.transaction() build = _require(state["pet_builds"], build_id, "build") pet = _require(state["pet_profiles"], build["pet_id"], "pet") media = [_require(state["pet_media"], media_id, "media") for media_id in build["input_media_ids"]] now = _now() build["status"] = "processing" build["updated_at"] = now self.store.write(state) result = self.build_runner(build=build.copy(), pet=pet.copy(), media=[item.copy() for item in media], output_root=self.output_root) state = self.store.transaction() build = _require(state["pet_builds"], build_id, "build") now = _now() _apply_build_success(build, result=result, output_root=self.output_root, now=now) _append_provider_call_metrics(state, build=build, result=result, now=now) _append_event(state, user_id=build["user_id"], pet_id=build["pet_id"], build_id=build_id, event_name="pet_build_completed", now=now) self.store.write(state) return { "build_id": build_id, "user_visible_status": "preview_ready", "preview_available": True, "updated_at": now, } def claim_next_build(self, *, worker_id: str) -> dict[str, Any]: state = self.store.transaction() item = find_next_queued_item(state) if item is None: raise KeyError("queued build not found") build = _require(state["pet_builds"], item["build_id"], "build") if build["status"] != "queued": item["status"] = "skipped" self.store.write(state) return self.claim_next_build(worker_id=worker_id) now = _now() item.update( { "status": "processing", "claimed_by": worker_id, "claimed_at": now, "updated_at": now, } ) build.update( { "status": "processing", "claimed_by": worker_id, "updated_at": now, } ) self.store.write(state) return { "build_id": build["build_id"], "user_visible_status": "processing", "claimed_at": now, } def complete_claimed_build(self, build_id: str, *, worker_id: str) -> dict[str, Any]: state = self.store.transaction() build = _require(state["pet_builds"], build_id, "build") if build["status"] != "processing" or build.get("claimed_by") != worker_id: raise StateTransitionError("build must be claimed by this worker before completion") pet = _require(state["pet_profiles"], build["pet_id"], "pet") media = [_require(state["pet_media"], media_id, "media") for media_id in build["input_media_ids"]] self.store.write(state) result = self.build_runner(build=build.copy(), pet=pet.copy(), media=[item.copy() for item in media], output_root=self.output_root) state = self.store.transaction() build = _require(state["pet_builds"], build_id, "build") now = _now() _apply_build_success(build, result=result, output_root=self.output_root, now=now) _append_provider_call_metrics(state, build=build, result=result, now=now) for item in state.setdefault("build_queue", []): if item.get("build_id") == build_id: item.update({"status": "completed", "updated_at": now, "completed_at": now}) _append_event(state, user_id=build["user_id"], pet_id=build["pet_id"], build_id=build_id, event_name="pet_build_completed", now=now) self.store.write(state) return { "build_id": build_id, "user_visible_status": "preview_ready", "preview_available": True, "updated_at": now, } def fail_claimed_build( self, build_id: str, *, worker_id: str, failure_code: str, failure_reason: str, ) -> dict[str, Any]: state = self.store.transaction() build = _require(state["pet_builds"], build_id, "build") if build["status"] != "processing" or build.get("claimed_by") != worker_id: raise StateTransitionError("build must be claimed by this worker before failure") now = _now() build.update( { "status": "failed", "failure_code": failure_code, "failure_reason": _safe_failure_reason(failure_reason), "updated_at": now, "completed_at": now, } ) for item in state.setdefault("build_queue", []): if item.get("build_id") == build_id: item.update({"status": "failed", "updated_at": now, "failure_code": failure_code}) _append_event(state, user_id=build["user_id"], pet_id=build["pet_id"], build_id=build_id, event_name="pet_build_failed", now=now) self.store.write(state) return { "build_id": build_id, "user_visible_status": "failed", "failure_code": failure_code, "updated_at": now, } def get_build(self, build_id: str) -> dict[str, Any]: state = self.store.read() build = _require(state["pet_builds"], build_id, "build") payload = { "build_id": build_id, "user_visible_status": _visible_build_status(build["status"]), "preview_available": build["status"] == "preview_ready", "updated_at": build["updated_at"], } if build["status"] == "failed": payload["failure_code"] = str(build.get("failure_code") or "worker_error") payload["failure_reason"] = _safe_failure_reason( str(build.get("failure_reason") or "Generation failed. Please try again.") ) return payload def get_provider_metrics( self, *, user_id: str, provider: str | None = None, build_id: str | None = None, ) -> dict[str, Any]: state = self.store.read() metrics = [ item for item in state.get("provider_call_metrics", []) if isinstance(item, dict) and item.get("user_id") == user_id and (not provider or item.get("provider") == provider) and (not build_id or item.get("build_id") == build_id) ] return summarize_provider_call_metrics(metrics, provider=provider, build_id=build_id) def get_build_preview(self, build_id: str) -> dict[str, Any]: state = self.store.read() build = _require(state["pet_builds"], build_id, "build") if build["status"] != "preview_ready": raise ValueError("preview is only available for preview_ready builds") return { "build_id": build_id, "preview_asset_url": f"/api/builds/{build_id}/preview/contact-sheet", "actions": list(build.get("actions") or ALLOWED_ACTIONS), "approval_required": True, } def get_build_preview_asset(self, build_id: str, *, user_id: str) -> dict[str, Any]: state = self.store.read() build = _require_owned_build(state, user_id=user_id, build_id=build_id) if build["status"] != "preview_ready": raise ValueError("preview is only available for preview_ready builds") build_dir_ref = str(build.get("build_dir_ref") or "") preview_ref = str(build.get("preview_ref") or "") if not build_dir_ref or Path(build_dir_ref).is_absolute() or ".." in Path(build_dir_ref).parts: raise StateTransitionError("preview build directory is not available") if not preview_ref or Path(preview_ref).is_absolute() or ".." in Path(preview_ref).parts: raise StateTransitionError("preview asset path is not safe") preview_path = self.output_root / build_dir_ref / preview_ref if not preview_path.exists(): raise FileNotFoundError("preview asset not found") preview_bytes = _preview_asset_bytes(preview_path=preview_path, build_root=self.output_root / build_dir_ref) return { "build_id": build_id, "user_visible_status": "preview_asset_ready", "filename": preview_path.name, "mime_type": _mime_type_for_asset(preview_path.name), "sha256": hashlib.sha256(preview_bytes).hexdigest(), "base64": base64.b64encode(preview_bytes).decode("ascii"), } def approve_build(self, build_id: str, *, user_id: str, approval_answers: dict[str, Any]) -> dict[str, Any]: state = self.store.transaction() build = _require_owned_build(state, user_id=user_id, build_id=build_id) assert_can_approve_build(build_status=build["status"], approval_answers=approval_answers) now = _now() build.update( { "status": "approved", "user_approved": True, "approval_answers": dict(approval_answers), "approved_at": now, "updated_at": now, } ) _append_event(state, user_id=user_id, pet_id=build["pet_id"], build_id=build_id, event_name="pet_build_approved", now=now) self.store.write(state) return { "build_id": build_id, "user_visible_status": "approved", "approved_at": now, } def create_package_download(self, build_id: str, *, user_id: str, entitlement_status: str | None = None) -> dict[str, Any]: state = self.store.transaction() build = _require_owned_build(state, user_id=user_id, build_id=build_id) now = _now() download_entitlement = _find_active_entitlement( state, user_id=user_id, pet_id=build["pet_id"], entitlement_type="package_download", now=now, ) effective_entitlement_status = str(download_entitlement.get("status") or "inactive") assert_can_mark_package_ready( build_status=build["status"], user_approved=bool(build["user_approved"]), entitlement_status=effective_entitlement_status, ) package_id = _next_id(state, "package", "pkg") actions = list(build.get("actions") or ALLOWED_ACTIONS) assert_can_download_package( package_status="approved_for_download", package_user_approved=True, entitlement_status=effective_entitlement_status, checksum_valid=bool(build.get("checksum_valid", True)), runtime_compatible=bool(build.get("runtime_compatible", True)), allowed_actions=actions, ) _consume_entitlement(state, download_entitlement, now=now, event_user_id=user_id, event_pet_id=build["pet_id"]) build["status"] = "package_ready" build["updated_at"] = now state["pet_packages"][package_id] = { "package_id": package_id, "build_id": build_id, "pet_id": build["pet_id"], "user_id": user_id, "status": "approved_for_download", "package_storage_ref": build["output_package_ref"], "allowed_actions": actions, "user_approved": True, "created_at": now, "updated_at": now, } _append_event(state, user_id=user_id, pet_id=build["pet_id"], build_id=build_id, event_name="pet_package_downloaded", now=now) self.store.write(state) return { "package_id": package_id, "build_id": build_id, "user_visible_status": "download_ready", "download_url": f"/api/packages/{package_id}/download", } def check_package_runtime_access( self, package_id: str, *, user_id: str, now: str | None = None, ) -> dict[str, Any]: state = self.store.read() package = _require(state["pet_packages"], package_id, "package") if package["user_id"] != user_id: raise PermissionError("package does not belong to user") effective_now = now or _now() runtime_entitlement = _find_active_entitlement( state, user_id=user_id, pet_id=package["pet_id"], entitlement_type="package_runtime_access", now=effective_now, ) assert_can_run_package( package_status=package["status"], package_user_approved=bool(package.get("user_approved")), runtime_entitlement_status=str(runtime_entitlement.get("status") or "inactive"), runtime_entitlement_expires_at=runtime_entitlement.get("expires_at"), now=effective_now, checksum_valid=bool(package.get("checksum_valid", True)), runtime_compatible=bool(package.get("runtime_compatible", True)), allowed_actions=list(package.get("allowed_actions") or ALLOWED_ACTIONS), ) return { "package_id": package_id, "user_visible_status": "runtime_access_granted", "expires_at": runtime_entitlement.get("expires_at"), } def create_test_checkout(self, *, user_id: str, build_id: str, product_code: str) -> dict[str, Any]: state = self.store.transaction() build = _require_owned_build(state, user_id=user_id, build_id=build_id) if build["status"] != "approved" or build.get("user_approved") is not True: raise StateTransitionError("approved build with user approval is required before checkout") now = _now() order_id = _next_id(state, "order", "ord") entitlement_id = _next_id(state, "entitlement", "ent") state.setdefault("orders", {})[order_id] = { "order_id": order_id, "user_id": user_id, "pet_id": build["pet_id"], "build_id": build_id, "provider": "manual", "status": "paid", "product_code": product_code, "currency": "USD", "amount_total_cents": 0, "created_at": now, "updated_at": now, "paid_at": now, } state.setdefault("entitlements", {})[entitlement_id] = { "entitlement_id": entitlement_id, "user_id": user_id, "order_id": order_id, "pet_id": build["pet_id"], "build_id": build_id, "type": "package_download", "status": "active", "quantity_total": 1, "quantity_used": 0, "created_at": now, "updated_at": now, } _append_event(state, user_id=user_id, pet_id=build["pet_id"], build_id=build_id, event_name="order_paid", now=now) _append_event(state, user_id=user_id, pet_id=build["pet_id"], build_id=build_id, event_name="entitlement_granted", now=now) self.store.write(state) return { "order_id": order_id, "entitlement_id": entitlement_id, "build_id": build_id, "user_visible_status": "paid", "created_at": now, } def delete_media(self, media_id: str, *, user_id: str) -> dict[str, Any]: state = self.store.transaction() media = _require(state["pet_media"], media_id, "media") if media["user_id"] != user_id: raise PermissionError("media does not belong to user") now = _now() media["deletion_status"] = "deleted" media["deleted_at"] = now media["updated_at"] = now if self.media_storage is not None: self.media_storage.delete(media["storage_ref"]) _append_event(state, user_id=user_id, pet_id=media["pet_id"], event_name="pet_media_deleted", now=now) self.store.write(state) return { "media_id": media_id, "deletion_status": "deleted", "updated_at": now, } def get_package_download(self, package_id: str, *, user_id: str) -> dict[str, Any]: state = self.store.read() package = _require(state["pet_packages"], package_id, "package") if package["user_id"] != user_id: raise PermissionError("package does not belong to user") if package["status"] != "approved_for_download": raise StateTransitionError("package is not available for download") return { "package_id": package_id, "build_id": package["build_id"], "user_visible_status": "download_ready", "download_url": f"/api/packages/{package_id}/download", } def get_package_bundle(self, package_id: str, *, user_id: str) -> dict[str, Any]: state = self.store.read() package = _require(state["pet_packages"], package_id, "package") if package["user_id"] != user_id: raise PermissionError("package does not belong to user") if package["status"] != "approved_for_download": raise StateTransitionError("package is not available for bundle export") build = _require(state["pet_builds"], package["build_id"], "build") build_dir_ref = str(build.get("build_dir_ref") or "") if not build_dir_ref or Path(build_dir_ref).is_absolute() or ".." in Path(build_dir_ref).parts: raise StateTransitionError("package build directory is not available") package_dir = self.output_root / build_dir_ref / "package" pet_json_path = package_dir / "pet.json" if not pet_json_path.exists(): raise FileNotFoundError("pet package manifest not found") manifest = json.loads(pet_json_path.read_text(encoding="utf-8")) manifest["userApproved"] = True spritesheet_name = _bundle_spritesheet_name(manifest=manifest, package_dir=package_dir) manifest["spritesheetPath"] = spritesheet_name manifest["spritesheet_path"] = spritesheet_name spritesheet_path = package_dir / spritesheet_name asset_bytes = spritesheet_path.read_bytes() return { "package_id": package_id, "build_id": package["build_id"], "user_visible_status": "bundle_ready", "manifest": manifest, "assets": [ { "filename": spritesheet_name, "mime_type": _mime_type_for_asset(spritesheet_name), "sha256": hashlib.sha256(asset_bytes).hexdigest(), "base64": base64.b64encode(asset_bytes).decode("ascii"), } ], } def delete_package(self, package_id: str, *, user_id: str) -> dict[str, Any]: state = self.store.transaction() package = _require(state["pet_packages"], package_id, "package") if package["user_id"] != user_id: raise PermissionError("package does not belong to user") now = _now() package["status"] = "deleted" package["deleted_at"] = now package["updated_at"] = now _append_event(state, user_id=user_id, pet_id=package["pet_id"], build_id=package["build_id"], event_name="pet_package_deleted", now=now) self.store.write(state) return { "package_id": package_id, "user_visible_status": "deleted", "updated_at": now, } def _now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def _one_month_later(value: str) -> str: base = datetime.fromisoformat(value.replace("Z", "+00:00")) return (base + timedelta(days=31)).isoformat().replace("+00:00", "Z") def _next_id(state: dict[str, Any], counter: str, prefix: str) -> str: state["counters"][counter] = int(state["counters"].get(counter, 0)) + 1 return f"{prefix}_{state['counters'][counter]:06d}" def _ensure_user(state: dict[str, Any], user_id: str, now: str) -> None: state["users"].setdefault( user_id, { "user_id": user_id, "status": "active", "created_at": now, "updated_at": now, }, ) def _ensure_default_price_plan(state: dict[str, Any], *, now: str, plan_code: str) -> None: state.setdefault("price_plans", {}).setdefault( plan_code, { "plan_code": plan_code, "provider": "stripe", "currency": "usd", "list_price_cents": 1699, "active_price_cents": 999, "billing_interval": "month", "included_build_count": 1, "included_revision_count": 3, "included_package_download_count": 1, "included_runtime_access": True, "model_cost_cap_cents": 500, "status": "active", "created_at": now, "updated_at": now, }, ) def _create_paid_order( state: dict[str, Any], *, user_id: str, pet_id: str | None, subscription_id: str, provider: str, product_code: str, amount_total_cents: int, now: str, ) -> str: order_id = _next_id(state, "order", "ord") state.setdefault("orders", {})[order_id] = { "order_id": order_id, "user_id": user_id, "pet_id": pet_id, "subscription_id": subscription_id, "provider": provider, "status": "paid", "product_code": product_code, "currency": "usd", "amount_total_cents": amount_total_cents, "created_at": now, "updated_at": now, "paid_at": now, } return order_id def _grant_monthly_entitlements( state: dict[str, Any], *, user_id: str, pet_id: str | None, subscription_id: str, order_id: str | None, starts_at: str, expires_at: str, now: str, ) -> list[str]: quantities = { "custom_pet_build": 1, "revision_credit": 3, "package_download": 1, "package_runtime_access": 1, } entitlement_ids = [] for entitlement_type in MONTHLY_ENTITLEMENT_TYPES: entitlement_id = _next_id(state, "entitlement", "ent") entitlement_ids.append(entitlement_id) state.setdefault("entitlements", {})[entitlement_id] = { "entitlement_id": entitlement_id, "user_id": user_id, "order_id": order_id, "subscription_id": subscription_id, "pet_id": pet_id, "type": entitlement_type, "status": "active", "quantity_total": quantities[entitlement_type], "quantity_used": 0, "starts_at": starts_at, "expires_at": expires_at, "created_at": now, "updated_at": now, } return entitlement_ids def _find_active_entitlement( state: dict[str, Any], *, user_id: str, entitlement_type: str, now: str, pet_id: str | None = None, ) -> dict[str, Any]: expired_candidate: dict[str, Any] | None = None for entitlement in state.setdefault("entitlements", {}).values(): if entitlement.get("user_id") != user_id or entitlement.get("type") != entitlement_type: continue if pet_id is not None and entitlement.get("pet_id") not in {None, pet_id}: continue if entitlement.get("expires_at") and str(entitlement["expires_at"]) <= now: expired_candidate = dict(entitlement) expired_candidate["status"] = "expired" continue if entitlement.get("status") == "active" and int(entitlement.get("quantity_used", 0)) < int( entitlement.get("quantity_total", 0) ): return entitlement if expired_candidate is not None: return expired_candidate return { "status": "inactive", "quantity_total": 0, "quantity_used": 0, "expires_at": None, } def _consume_entitlement( state: dict[str, Any], entitlement: dict[str, Any], *, now: str, event_user_id: str, event_pet_id: str | None, ) -> None: entitlement_id = entitlement.get("entitlement_id") if not entitlement_id: raise StateTransitionError("active entitlement is required") entitlement["quantity_used"] = int(entitlement.get("quantity_used", 0)) + 1 if int(entitlement["quantity_used"]) >= int(entitlement.get("quantity_total", 0)): entitlement["status"] = "consumed" entitlement["consumed_at"] = now entitlement["updated_at"] = now _append_event( state, user_id=event_user_id, pet_id=event_pet_id, event_name="entitlement_consumed", now=now, ) def _get_or_create_model_cost_quota(state: dict[str, Any], *, user_id: str, pet_id: str, now: str) -> dict[str, Any]: for quota in state.setdefault("quota_buckets", {}).values(): if quota.get("user_id") == user_id and quota.get("pet_id") == pet_id and quota.get("quota_type") == "model_cost_cents": return quota quota_id = _next_id(state, "quota", "quo") state["quota_buckets"][quota_id] = { "quota_id": quota_id, "user_id": user_id, "pet_id": pet_id, "quota_type": "model_cost_cents", "limit_value": 500, "used_value": 0, "reserved_value": 0, "status": "active", "created_at": now, "updated_at": now, } return state["quota_buckets"][quota_id] def _hash_invite_code(code: str) -> str: return hashlib.sha256(code.strip().encode("utf-8")).hexdigest() def _find_invite_by_code(state: dict[str, Any], *, code: str, now: str) -> dict[str, Any]: code_hash = _hash_invite_code(code) for invite in state.setdefault("beta_invites", {}).values(): if invite.get("code_hash") != code_hash: continue if invite.get("expires_at") and str(invite["expires_at"]) <= now: invite["status"] = "expired" raise StateTransitionError("invite has expired") if invite.get("status") != "active": raise StateTransitionError("invite is not active") return invite raise StateTransitionError("invite is not active") def _require(collection: dict[str, Any], key: str, label: str) -> dict[str, Any]: item = collection.get(key) if not isinstance(item, dict): raise KeyError(f"{label} not found") return item def _require_owned_pet(state: dict[str, Any], *, user_id: str, pet_id: str) -> dict[str, Any]: pet = _require(state["pet_profiles"], pet_id, "pet") if pet["user_id"] != user_id: raise PermissionError("pet does not belong to user") return pet def _require_owned_media(state: dict[str, Any], *, user_id: str, pet_id: str, media_id: str) -> dict[str, Any]: media = _require(state["pet_media"], media_id, "media") if media["user_id"] != user_id or media["pet_id"] != pet_id: raise PermissionError("media does not belong to pet") return media def _require_owned_build(state: dict[str, Any], *, user_id: str, build_id: str) -> dict[str, Any]: build = _require(state["pet_builds"], build_id, "build") if build["user_id"] != user_id: raise PermissionError("build does not belong to user") return build def _append_event( state: dict[str, Any], *, user_id: str, event_name: str, now: str, pet_id: str | None = None, build_id: str | None = None, ) -> None: event_id = _next_id(state, "event", "evt") state["events"].append( { "event_id": event_id, "user_id": user_id, "pet_id": pet_id, "build_id": build_id, "event_name": event_name, "event_properties": {}, "occurred_at": now, } ) def _visible_build_status(status: str) -> str: return { "queued": "queued", "processing": "processing", "preview_ready": "preview_ready", "approved": "approved", "package_ready": "download_ready", "failed": "failed", }.get(status, "draft") def _bundle_spritesheet_name(*, manifest: dict[str, Any], package_dir: Path) -> str: preferred_png = package_dir / "spritesheet.png" if preferred_png.exists(): return "spritesheet.png" spritesheet_name = str(manifest.get("spritesheetPath") or manifest.get("spritesheet_path") or "spritesheet.png") if Path(spritesheet_name).is_absolute() or ".." in Path(spritesheet_name).parts: raise StateTransitionError("package asset path is not safe") if not (package_dir / spritesheet_name).exists(): raise FileNotFoundError("pet package spritesheet not found") return spritesheet_name def _mime_type_for_asset(filename: str) -> str: suffix = Path(filename).suffix.lower() if suffix == ".png": return "image/png" if suffix == ".webp": return "image/webp" if suffix in {".jpg", ".jpeg"}: return "image/jpeg" if suffix == ".html": return "text/html" return "application/octet-stream" def _preview_asset_bytes(*, preview_path: Path, build_root: Path) -> bytes: if preview_path.suffix.lower() != ".html": return preview_path.read_bytes() html = preview_path.read_text(encoding="utf-8") def replace_src(match: re.Match[str]) -> str: quote = match.group("quote") src = match.group("src") if src.startswith(("data:", "http://", "https://", "/")): return match.group(0) asset_ref = Path(src) if asset_ref.is_absolute(): return match.group(0) asset_path = (preview_path.parent / asset_ref).resolve() try: asset_path.relative_to(build_root.resolve()) except ValueError: return match.group(0) if not asset_path.exists(): return match.group(0) mime_type = _mime_type_for_asset(asset_path.name) encoded = base64.b64encode(asset_path.read_bytes()).decode("ascii") return f'src={quote}data:{mime_type};base64,{encoded}{quote}' return re.sub(r"src=(?P[\"'])(?P[^\"']+)(?P=quote)", replace_src, html).encode("utf-8") def _safe_relative_ref(value: Any, root: Path) -> str: if value is None: return "" path = Path(value) try: return str(path.resolve().relative_to(root.resolve())) except (OSError, ValueError): return path.name def _entitlement_status_for_build(state: dict[str, Any], *, user_id: str, build_id: str) -> str: for entitlement in state.setdefault("entitlements", {}).values(): if ( entitlement.get("user_id") == user_id and entitlement.get("build_id") == build_id and entitlement.get("status") == "active" ): return "active" return "inactive" def _apply_build_success(build: dict[str, Any], *, result: dict[str, Any], output_root: Path, now: str) -> None: if result.get("validation_ok") is False: raise StateTransitionError("generated package validation failed") build.update( { "status": "preview_ready", "output_package_ref": str(result["package_ref"]), "preview_ref": str(result["preview_ref"]), "actions": list(result.get("actions") or ALLOWED_ACTIONS), "checksum_valid": bool(result.get("checksum_valid", True)), "runtime_compatible": bool(result.get("runtime_compatible", True)), "build_dir_ref": _safe_relative_ref(result.get("build_dir"), output_root), "updated_at": now, "completed_at": now, } ) def _append_provider_call_metrics( state: dict[str, Any], *, build: dict[str, Any], result: dict[str, Any], now: str, ) -> None: metrics = sanitize_provider_call_metrics( result.get("provider_call_metrics"), user_id=str(build.get("user_id") or ""), pet_id=str(build.get("pet_id") or ""), build_id=str(build.get("build_id") or ""), created_at=now, ) if metrics: state.setdefault("provider_call_metrics", []).extend(metrics) def _safe_failure_reason(reason: str) -> str: if "/" in reason or "\\" in reason or "traceback" in reason.lower() or "stack" in reason.lower(): return "Generation failed. Please try again." return reason[:220] def _assert_supported_image_upload(*, filename: str, content_type: str, content: bytes) -> None: if not content: raise ValueError("uploaded file is empty") if len(content) > MAX_UPLOAD_BYTES: raise ValueError("uploaded file is too large") if content_type not in {"image/png", "image/jpeg", "image/webp", "image/gif"}: raise ValueError("uploaded file must be a supported image") suffix = Path(filename).suffix.lower() if suffix and suffix not in {".png", ".jpg", ".jpeg", ".webp", ".gif"}: raise ValueError("uploaded filename must use an image extension") if content_type == "image/png" and not content.startswith(b"\x89PNG\r\n\x1a\n"): raise ValueError("uploaded PNG is not readable") if content_type == "image/jpeg" and not content.startswith(b"\xff\xd8"): raise ValueError("uploaded JPEG is not readable") if content_type == "image/webp" and not (content.startswith(b"RIFF") and content[8:12] == b"WEBP"): raise ValueError("uploaded WebP is not readable") if content_type == "image/gif" and not (content.startswith(b"GIF87a") or content.startswith(b"GIF89a")): raise ValueError("uploaded GIF is not readable")