#!/usr/bin/env python3 """ Art DAG L2 Server - ActivityPub Manages ownership registry, activities, and federation. - Registry of owned assets - ActivityPub actor endpoints - Sign and publish Create activities - Federation with other servers """ import hashlib import json import os import uuid from contextlib import asynccontextmanager from datetime import datetime, timezone from pathlib import Path from typing import Optional from urllib.parse import urlparse from fastapi import FastAPI, HTTPException, Request, Response, Depends, Cookie from fastapi.responses import JSONResponse, HTMLResponse, RedirectResponse from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel import requests import markdown import db from auth import ( UserCreate, UserLogin, Token, User, create_user, authenticate_user, create_access_token, verify_token, get_current_user ) # Configuration DOMAIN = os.environ.get("ARTDAG_DOMAIN", "artdag.rose-ash.com") DATA_DIR = Path(os.environ.get("ARTDAG_DATA", str(Path.home() / ".artdag" / "l2"))) L1_PUBLIC_URL = os.environ.get("L1_PUBLIC_URL", "https://celery-artdag.rose-ash.com") EFFECTS_REPO_URL = os.environ.get("EFFECTS_REPO_URL", "https://git.rose-ash.com/art-dag/effects") # Ensure data directory exists DATA_DIR.mkdir(parents=True, exist_ok=True) (DATA_DIR / "assets").mkdir(exist_ok=True) # Load README README_PATH = Path(__file__).parent / "README.md" README_CONTENT = "" if README_PATH.exists(): README_CONTENT = README_PATH.read_text() @asynccontextmanager async def lifespan(app: FastAPI): """Manage database connection pool lifecycle.""" await db.init_pool() yield await db.close_pool() app = FastAPI( title="Art DAG L2 Server", description="ActivityPub server for Art DAG ownership and federation", version="0.1.0", lifespan=lifespan ) # ============ Data Models ============ class Asset(BaseModel): """An owned asset.""" name: str content_hash: str asset_type: str # image, video, effect, recipe, infrastructure tags: list[str] = [] metadata: dict = {} url: Optional[str] = None provenance: Optional[dict] = None created_at: str = "" class Activity(BaseModel): """An ActivityPub activity.""" activity_id: str activity_type: str # Create, Update, Delete, Announce actor_id: str object_data: dict published: str signature: Optional[dict] = None class RegisterRequest(BaseModel): """Request to register an asset.""" name: str content_hash: str asset_type: str tags: list[str] = [] metadata: dict = {} url: Optional[str] = None provenance: Optional[dict] = None class RecordRunRequest(BaseModel): """Request to record an L1 run.""" run_id: str output_name: str l1_server: str # URL of the L1 server that has this run class PublishCacheRequest(BaseModel): """Request to publish a cache item from L1.""" content_hash: str asset_name: str asset_type: str = "image" origin: dict # {type: "self"|"external", url?: str, note?: str} description: Optional[str] = None tags: list[str] = [] metadata: dict = {} class UpdateAssetRequest(BaseModel): """Request to update an existing asset.""" description: Optional[str] = None tags: Optional[list[str]] = None metadata: Optional[dict] = None origin: Optional[dict] = None # ============ Storage (Database) ============ async def load_registry() -> dict: """Load registry from database.""" assets = await db.get_all_assets() return {"version": "1.0", "assets": assets} async def load_activities() -> list: """Load activities from database.""" return await db.get_all_activities() def load_actor(username: str) -> dict: """Load actor data for a specific user with public key if available.""" actor = { "id": f"https://{DOMAIN}/users/{username}", "type": "Person", "preferredUsername": username, "name": username, "inbox": f"https://{DOMAIN}/users/{username}/inbox", "outbox": f"https://{DOMAIN}/users/{username}/outbox", "followers": f"https://{DOMAIN}/users/{username}/followers", "following": f"https://{DOMAIN}/users/{username}/following", } # Add public key if available from keys import has_keys, load_public_key_pem if has_keys(DATA_DIR, username): actor["publicKey"] = { "id": f"https://{DOMAIN}/users/{username}#main-key", "owner": f"https://{DOMAIN}/users/{username}", "publicKeyPem": load_public_key_pem(DATA_DIR, username) } return actor async def user_exists(username: str) -> bool: """Check if a user exists.""" return await db.user_exists(username) async def load_followers() -> list: """Load followers list from database.""" return await db.get_all_followers() # ============ Signing ============ from keys import has_keys, load_public_key_pem, create_signature def sign_activity(activity: dict, username: str) -> dict: """Sign an activity with the user's RSA private key.""" if not has_keys(DATA_DIR, username): # No keys - use placeholder (for testing) activity["signature"] = { "type": "RsaSignature2017", "creator": f"https://{DOMAIN}/users/{username}#main-key", "created": datetime.now(timezone.utc).isoformat(), "signatureValue": "NO_KEYS_CONFIGURED" } else: activity["signature"] = create_signature(DATA_DIR, username, DOMAIN, activity) return activity # ============ HTML Templates ============ # Tailwind CSS config for L2 - dark theme to match L1 TAILWIND_CONFIG = ''' ''' def base_html(title: str, content: str, username: str = None) -> str: """Base HTML template with Tailwind CSS dark theme.""" user_section = f'''
Logged in as {username} Logout
''' if username else '''
Login | Register
''' return f''' {title} - Art DAG L2 {TAILWIND_CONFIG}

Art DAG L2

{user_section}
{content}
''' def get_user_from_cookie(request: Request) -> Optional[str]: """Get username from auth cookie.""" token = request.cookies.get("auth_token") if token: return verify_token(token) return None def wants_html(request: Request) -> bool: """Check if request wants HTML (browser) vs JSON (API).""" accept = request.headers.get("accept", "") return "text/html" in accept and "application/json" not in accept and "application/activity+json" not in accept # ============ Auth UI Endpoints ============ @app.get("/login", response_class=HTMLResponse) async def ui_login_page(request: Request): """Login page.""" username = get_user_from_cookie(request) if username: return HTMLResponse(base_html("Already Logged In", f'''
You are already logged in as {username}

Go to home page

''', username)) content = '''

Login

Don't have an account? Register

''' return HTMLResponse(base_html("Login", content)) @app.post("/login", response_class=HTMLResponse) async def ui_login_submit(request: Request): """Handle login form submission.""" form = await request.form() username = form.get("username", "").strip() password = form.get("password", "") if not username or not password: return HTMLResponse('
Username and password are required
') user = await authenticate_user(DATA_DIR, username, password) if not user: return HTMLResponse('
Invalid username or password
') token = create_access_token(user.username) response = HTMLResponse(f'''
Login successful! Redirecting...
''') response.set_cookie( key="auth_token", value=token.access_token, httponly=True, max_age=60 * 60 * 24 * 30, # 30 days samesite="lax" ) return response @app.get("/register", response_class=HTMLResponse) async def ui_register_page(request: Request): """Register page.""" username = get_user_from_cookie(request) if username: return HTMLResponse(base_html("Already Logged In", f'''
You are already logged in as {username}

Go to home page

''', username)) content = '''

Register

Already have an account? Login

''' return HTMLResponse(base_html("Register", content)) @app.post("/register", response_class=HTMLResponse) async def ui_register_submit(request: Request): """Handle register form submission.""" form = await request.form() username = form.get("username", "").strip() email = form.get("email", "").strip() or None password = form.get("password", "") password2 = form.get("password2", "") if not username or not password: return HTMLResponse('
Username and password are required
') if password != password2: return HTMLResponse('
Passwords do not match
') if len(password) < 6: return HTMLResponse('
Password must be at least 6 characters
') try: user = await create_user(DATA_DIR, username, password, email) except ValueError as e: return HTMLResponse(f'
{str(e)}
') token = create_access_token(user.username) response = HTMLResponse(f'''
Registration successful! Redirecting...
''') response.set_cookie( key="auth_token", value=token.access_token, httponly=True, max_age=60 * 60 * 24 * 30, # 30 days samesite="lax" ) return response @app.get("/logout") async def logout(): """Handle logout - clear cookie and redirect to home.""" response = RedirectResponse(url="/", status_code=302) response.delete_cookie("auth_token") return response # ============ HTML Rendering Helpers ============ async def ui_activity_detail(activity_index: int, request: Request): """Activity detail page with full content display. Helper function for HTML rendering.""" username = get_user_from_cookie(request) activities = await load_activities() if activity_index < 0 or activity_index >= len(activities): content = '''

Activity Not Found

This activity does not exist.

← Back to Activities

''' return HTMLResponse(base_html("Activity Not Found", content, username)) activity = activities[activity_index] activity_type = activity.get("activity_type", "") activity_id = activity.get("activity_id", "") actor_id = activity.get("actor_id", "") actor_name = actor_id.split("/")[-1] if actor_id else "unknown" published = activity.get("published", "")[:10] obj = activity.get("object_data", {}) # Object details obj_name = obj.get("name", "Untitled") obj_type = obj.get("type", "") content_hash_obj = obj.get("contentHash", {}) content_hash = content_hash_obj.get("value", "") if isinstance(content_hash_obj, dict) else "" media_type = obj.get("mediaType", "") description = obj.get("summary", "") or obj.get("content", "") # Provenance from object - or fallback to registry asset provenance = obj.get("provenance", {}) origin = obj.get("origin", {}) # Fallback: if activity doesn't have provenance, look up the asset from registry if not provenance or not origin: registry = await load_registry() assets = registry.get("assets", {}) # Find asset by content_hash or name for asset_name, asset_data in assets.items(): if asset_data.get("content_hash") == content_hash or asset_data.get("name") == obj_name: if not provenance: provenance = asset_data.get("provenance", {}) if not origin: origin = asset_data.get("origin", {}) break # Type colors type_color = "bg-green-600" if activity_type == "Create" else "bg-yellow-600" if activity_type == "Update" else "bg-gray-600" obj_type_color = "bg-blue-600" if "Image" in obj_type else "bg-purple-600" if "Video" in obj_type else "bg-gray-600" # Determine L1 server and asset type l1_server = provenance.get("l1_server", L1_PUBLIC_URL).rstrip("/") if provenance else L1_PUBLIC_URL.rstrip("/") is_video = "Video" in obj_type or "video" in media_type # Content display if is_video: content_html = f'''
Download Original
''' elif "Image" in obj_type or "image" in media_type: content_html = f'''
{obj_name}
Download
''' else: content_html = f'''

Content type: {media_type or obj_type}

Download
''' # Origin display origin_html = 'Not specified' if origin: origin_type = origin.get("type", "") if origin_type == "self": origin_html = 'Original content by author' elif origin_type == "external": origin_url = origin.get("url", "") origin_note = origin.get("note", "") origin_html = f'{origin_url}' if origin_note: origin_html += f'

{origin_note}

' # Provenance section provenance_html = "" if provenance and provenance.get("recipe"): recipe = provenance.get("recipe", "") inputs = provenance.get("inputs", []) l1_run_id = provenance.get("l1_run_id", "") rendered_at = provenance.get("rendered_at", "")[:10] if provenance.get("rendered_at") else "" effects_commit = provenance.get("effects_commit", "") effect_url = provenance.get("effect_url") infrastructure = provenance.get("infrastructure", {}) if not effect_url: if effects_commit and effects_commit != "unknown": effect_url = f"{EFFECTS_REPO_URL}/src/commit/{effects_commit}/{recipe}" else: effect_url = f"{EFFECTS_REPO_URL}/src/branch/main/{recipe}" # Build inputs display - show actual content as thumbnails inputs_html = "" for inp in inputs: inp_hash = inp.get("content_hash", "") if isinstance(inp, dict) else inp if inp_hash: inputs_html += f'''
Input
{inp_hash[:16]}... view
''' # Infrastructure display infra_html = "" if infrastructure: software = infrastructure.get("software", {}) hardware = infrastructure.get("hardware", {}) if software or hardware: infra_parts = [] if software: infra_parts.append(f"Software: {software.get('name', 'unknown')}") if hardware: infra_parts.append(f"Hardware: {hardware.get('name', 'unknown')}") infra_html = f'

{" | ".join(infra_parts)}

' provenance_html = f'''

Provenance

This content was created by applying an effect to input content.

Effect

{recipe} {f'
Commit: {effects_commit[:12]}...
' if effects_commit else ''}

Input(s)

{inputs_html if inputs_html else 'No inputs recorded'}

Rendered

{rendered_at if rendered_at else 'Unknown'} {infra_html}
''' content = f'''

← Back to Activities

{activity_type}

{obj_name}

{obj_type}
{content_html}

Actor

{actor_name}

Description

{description if description else 'No description'}

Origin

{origin_html}

Content Hash

{content_hash}

Published

{published}

Activity ID

{activity_id}
{provenance_html}

ActivityPub

Object URL: https://{DOMAIN}/objects/{content_hash}

Actor: {actor_id}

''' return HTMLResponse(base_html(f"Activity: {obj_name}", content, username)) async def ui_asset_detail(name: str, request: Request): """Asset detail page with content preview and provenance. Helper function for HTML rendering.""" username = get_user_from_cookie(request) registry = await load_registry() assets = registry.get("assets", {}) if name not in assets: content = f'''

Asset Not Found

No asset named "{name}" exists.

← Back to Assets

''' return HTMLResponse(base_html("Asset Not Found", content, username)) asset = assets[name] owner = asset.get("owner", "unknown") content_hash = asset.get("content_hash", "") asset_type = asset.get("asset_type", "") tags = asset.get("tags", []) description = asset.get("description", "") origin = asset.get("origin", {}) provenance = asset.get("provenance", {}) metadata = asset.get("metadata", {}) created_at = asset.get("created_at", "")[:10] type_color = "bg-blue-600" if asset_type == "image" else "bg-purple-600" if asset_type == "video" else "bg-gray-600" # Determine L1 server URL for content l1_server = provenance.get("l1_server", L1_PUBLIC_URL).rstrip("/") # Content display - image or video from L1 if asset_type == "video": # Use iOS-compatible MP4 endpoint content_html = f'''
Download Original
''' elif asset_type == "image": content_html = f'''
{name}
Download
''' else: content_html = f'''

Content type: {asset_type}

Download
''' # Origin display origin_html = 'Not specified' if origin: origin_type = origin.get("type", "unknown") if origin_type == "self": origin_html = 'Original content by author' elif origin_type == "external": origin_url = origin.get("url", "") origin_note = origin.get("note", "") origin_html = f'{origin_url}' if origin_note: origin_html += f'

{origin_note}

' # Tags display tags_html = 'No tags' if tags: tags_html = " ".join([f'{t}' for t in tags]) # Provenance section - for rendered outputs provenance_html = "" if provenance: recipe = provenance.get("recipe", "") inputs = provenance.get("inputs", []) l1_run_id = provenance.get("l1_run_id", "") rendered_at = provenance.get("rendered_at", "")[:10] if provenance.get("rendered_at") else "" effects_commit = provenance.get("effects_commit", "") infrastructure = provenance.get("infrastructure", {}) # Use stored effect_url or build fallback effect_url = provenance.get("effect_url") if not effect_url: # Fallback for older records if effects_commit and effects_commit != "unknown": effect_url = f"{EFFECTS_REPO_URL}/src/commit/{effects_commit}/{recipe}" else: effect_url = f"{EFFECTS_REPO_URL}/src/branch/main/{recipe}" # Build inputs display - show actual content as thumbnails inputs_html = "" for inp in inputs: inp_hash = inp.get("content_hash", "") if isinstance(inp, dict) else inp if inp_hash: inputs_html += f'''
Input
{inp_hash[:16]}... view
''' # Infrastructure display infra_html = "" if infrastructure: software = infrastructure.get("software", {}) hardware = infrastructure.get("hardware", {}) if software or hardware: infra_html = f'''

Infrastructure

{f"Software: {software.get('name', 'unknown')}" if software else ""} {f" ({software.get('content_hash', '')[:16]}...)" if software.get('content_hash') else ""} {" | " if software and hardware else ""} {f"Hardware: {hardware.get('name', 'unknown')}" if hardware else ""}
''' provenance_html = f'''

Provenance

This asset was created by applying an effect to input content.

Effect

{recipe} {f'
Commit: {effects_commit[:12]}...
' if effects_commit else ''}

Input(s)

{inputs_html if inputs_html else 'No inputs recorded'}

Rendered

{rendered_at if rendered_at else 'Unknown'}
{infra_html}
''' content = f'''

← Back to Assets

{name}

{asset_type}
{content_html}

Owner

{owner}

Description

{description if description else 'No description'}

Origin

{origin_html}

Content Hash

{content_hash}

Created

{created_at}

Tags

{tags_html}
{provenance_html}

ActivityPub

Object URL: https://{DOMAIN}/objects/{content_hash}

Owner Actor: https://{DOMAIN}/users/{owner}

''' return HTMLResponse(base_html(f"Asset: {name}", content, username)) async def ui_user_detail(username: str, request: Request): """User detail page showing their published assets. Helper function for HTML rendering.""" current_user = get_user_from_cookie(request) if not await user_exists(username): content = f'''

User Not Found

No user named "{username}" exists.

← Back to Users

''' return HTMLResponse(base_html("User Not Found", content, current_user)) # Get user's assets registry = await load_registry() all_assets = registry.get("assets", {}) user_assets = {name: asset for name, asset in all_assets.items() if asset.get("owner") == username} # Get user's activities all_activities = await load_activities() actor_id = f"https://{DOMAIN}/users/{username}" user_activities = [a for a in all_activities if a.get("actor_id") == actor_id] webfinger = f"@{username}@{DOMAIN}" # Assets table if user_assets: rows = "" for name, asset in sorted(user_assets.items(), key=lambda x: x[1].get("created_at", ""), reverse=True): hash_short = asset.get("content_hash", "")[:16] + "..." asset_type = asset.get("asset_type", "") type_color = "bg-blue-600" if asset_type == "image" else "bg-purple-600" if asset_type == "video" else "bg-gray-600" rows += f''' {name} {asset_type} {hash_short} {", ".join(asset.get("tags", []))} ''' assets_html = f'''
{rows}
Name Type Content Hash Tags
''' else: assets_html = '

No published assets yet.

' content = f'''

← Back to Users

{username}

{webfinger}
{len(user_assets)}
Published Assets
{len(user_activities)}
Activities

ActivityPub

Actor URL: https://{DOMAIN}/users/{username}

Outbox: https://{DOMAIN}/users/{username}/outbox

Published Assets ({len(user_assets)})

{assets_html} ''' return HTMLResponse(base_html(f"User: {username}", content, current_user)) # ============ API Endpoints ============ @app.get("/") async def root(request: Request): """Server info. HTML shows home page with counts, JSON returns stats.""" registry = await load_registry() activities = await load_activities() users = await db.get_all_users() assets_count = len(registry.get("assets", {})) activities_count = len(activities) users_count = len(users) if wants_html(request): username = get_user_from_cookie(request) readme_html = markdown.markdown(README_CONTENT, extensions=['tables', 'fenced_code']) content = f'''
{assets_count}
Assets
{activities_count}
Activities
{users_count}
Users
{readme_html}
''' return HTMLResponse(base_html("Home", content, username)) return { "name": "Art DAG L2 Server", "version": "0.1.0", "domain": DOMAIN, "assets_count": assets_count, "activities_count": activities_count, "users_count": users_count } # ============ Auth Endpoints ============ security = HTTPBearer(auto_error=False) async def get_optional_user( credentials: HTTPAuthorizationCredentials = Depends(security) ) -> Optional[User]: """Get current user if authenticated, None otherwise.""" if not credentials: return None return await get_current_user(DATA_DIR, credentials.credentials) async def get_required_user( credentials: HTTPAuthorizationCredentials = Depends(security) ) -> User: """Get current user, raise 401 if not authenticated.""" if not credentials: raise HTTPException(401, "Not authenticated") user = await get_current_user(DATA_DIR, credentials.credentials) if not user: raise HTTPException(401, "Invalid token") return user @app.post("/auth/register", response_model=Token) async def register(req: UserCreate): """Register a new user.""" try: user = await create_user(DATA_DIR, req.username, req.password, req.email) except ValueError as e: raise HTTPException(400, str(e)) return create_access_token(user.username) @app.post("/auth/login", response_model=Token) async def login(req: UserLogin): """Login and get access token.""" user = await authenticate_user(DATA_DIR, req.username, req.password) if not user: raise HTTPException(401, "Invalid username or password") return create_access_token(user.username) @app.get("/auth/me") async def get_me(user: User = Depends(get_required_user)): """Get current user info.""" return { "username": user.username, "email": user.email, "created_at": user.created_at } @app.post("/auth/verify") async def verify_auth(credentials: HTTPAuthorizationCredentials = Depends(security)): """Verify a token and return username. Used by L1 server.""" if not credentials: raise HTTPException(401, "No token provided") username = verify_token(credentials.credentials) if not username: raise HTTPException(401, "Invalid token") return {"username": username, "valid": True} @app.get("/.well-known/webfinger") async def webfinger(resource: str): """WebFinger endpoint for actor discovery.""" # Parse acct:username@domain if not resource.startswith("acct:"): raise HTTPException(400, "Resource must be acct: URI") acct = resource[5:] # Remove "acct:" if "@" not in acct: raise HTTPException(400, "Invalid acct format") username, domain = acct.split("@", 1) if domain != DOMAIN: raise HTTPException(404, f"Unknown domain: {domain}") if not await user_exists(username): raise HTTPException(404, f"Unknown user: {username}") return JSONResponse( content={ "subject": resource, "links": [ { "rel": "self", "type": "application/activity+json", "href": f"https://{DOMAIN}/users/{username}" } ] }, media_type="application/jrd+json" ) @app.get("/users") async def get_users_list(request: Request, page: int = 1, limit: int = 20): """Get all users. HTML for browsers (with infinite scroll), JSON for APIs (with pagination).""" all_users = list((await db.get_all_users()).items()) total = len(all_users) # Sort by username all_users.sort(key=lambda x: x[0]) # Pagination start = (page - 1) * limit end = start + limit users_page = all_users[start:end] has_more = end < total if wants_html(request): username = get_user_from_cookie(request) if not users_page: if page == 1: content = '''

Users

No users registered yet.

''' else: return HTMLResponse("") # Empty for infinite scroll else: rows = "" for uname, user_data in users_page: webfinger = f"@{uname}@{DOMAIN}" rows += f''' {uname} {webfinger} {user_data.get("created_at", "")[:10]} ''' # For infinite scroll, just return rows if not first page if page > 1: if has_more: rows += f''' Loading more... ''' return HTMLResponse(rows) # First page - full content infinite_scroll_trigger = "" if has_more: infinite_scroll_trigger = f''' Loading more... ''' content = f'''

Users ({total} total)

{rows} {infinite_scroll_trigger}
Username WebFinger Created
''' return HTMLResponse(base_html("Users", content, username)) # JSON response for APIs return { "users": [{"username": uname, **data} for uname, data in users_page], "pagination": { "page": page, "limit": limit, "total": total, "has_more": has_more } } @app.get("/users/{username}") async def get_actor(username: str, request: Request): """Get actor profile for any registered user. Content negotiation: HTML for browsers, JSON for APIs.""" if not await user_exists(username): if wants_html(request): content = f'''

User Not Found

No user named "{username}" exists.

← Back to Users

''' return HTMLResponse(base_html("User Not Found", content, get_user_from_cookie(request))) raise HTTPException(404, f"Unknown user: {username}") if wants_html(request): # Render user detail page return await ui_user_detail(username, request) actor = load_actor(username) # Add ActivityPub context actor["@context"] = [ "https://www.w3.org/ns/activitystreams", "https://w3id.org/security/v1" ] return JSONResponse( content=actor, media_type="application/activity+json" ) @app.get("/users/{username}/outbox") async def get_outbox(username: str, page: bool = False): """Get actor's outbox (activities they created).""" if not await user_exists(username): raise HTTPException(404, f"Unknown user: {username}") # Filter activities by this user's actor_id all_activities = await load_activities() actor_id = f"https://{DOMAIN}/users/{username}" user_activities = [a for a in all_activities if a.get("actor_id") == actor_id] if not page: return JSONResponse( content={ "@context": "https://www.w3.org/ns/activitystreams", "id": f"https://{DOMAIN}/users/{username}/outbox", "type": "OrderedCollection", "totalItems": len(user_activities), "first": f"https://{DOMAIN}/users/{username}/outbox?page=true" }, media_type="application/activity+json" ) # Return activities page return JSONResponse( content={ "@context": "https://www.w3.org/ns/activitystreams", "id": f"https://{DOMAIN}/users/{username}/outbox?page=true", "type": "OrderedCollectionPage", "partOf": f"https://{DOMAIN}/users/{username}/outbox", "orderedItems": user_activities }, media_type="application/activity+json" ) @app.post("/users/{username}/inbox") async def post_inbox(username: str, request: Request): """Receive activities from other servers.""" if not await user_exists(username): raise HTTPException(404, f"Unknown user: {username}") body = await request.json() activity_type = body.get("type") # Handle Follow requests if activity_type == "Follow": follower_url = body.get("actor") # Add follower to database await db.add_follower(username, follower_url, follower_url) # Send Accept (in production, do this async) # For now just acknowledge return {"status": "accepted"} # Handle other activity types return {"status": "received"} @app.get("/users/{username}/followers") async def get_followers(username: str): """Get actor's followers.""" if not await user_exists(username): raise HTTPException(404, f"Unknown user: {username}") # TODO: Per-user followers - for now use global followers followers = await load_followers() return JSONResponse( content={ "@context": "https://www.w3.org/ns/activitystreams", "id": f"https://{DOMAIN}/users/{username}/followers", "type": "OrderedCollection", "totalItems": len(followers), "orderedItems": followers }, media_type="application/activity+json" ) # ============ Assets Endpoints ============ @app.get("/assets") async def get_registry(request: Request, page: int = 1, limit: int = 20): """Get registry. HTML for browsers (with infinite scroll), JSON for APIs (with pagination).""" registry = await load_registry() all_assets = list(registry.get("assets", {}).items()) total = len(all_assets) # Sort by created_at descending all_assets.sort(key=lambda x: x[1].get("created_at", ""), reverse=True) # Pagination start = (page - 1) * limit end = start + limit assets_page = all_assets[start:end] has_more = end < total if wants_html(request): username = get_user_from_cookie(request) if not assets_page: if page == 1: content = '''

Registry

No assets registered yet.

''' else: return HTMLResponse("") # Empty for infinite scroll else: rows = "" for name, asset in assets_page: asset_type = asset.get("asset_type", "") type_color = "bg-blue-600" if asset_type == "image" else "bg-purple-600" if asset_type == "video" else "bg-gray-600" owner = asset.get("owner", "unknown") content_hash = asset.get("content_hash", "")[:16] + "..." rows += f''' {name} {asset_type} {owner} {content_hash} View ''' # For infinite scroll, just return rows if not first page if page > 1: if has_more: rows += f''' Loading more... ''' return HTMLResponse(rows) # First page - full content infinite_scroll_trigger = "" if has_more: infinite_scroll_trigger = f''' Loading more... ''' content = f'''

Registry ({total} assets)

{rows} {infinite_scroll_trigger}
Name Type Owner Hash
''' return HTMLResponse(base_html("Registry", content, username)) # JSON response for APIs return { "assets": {name: asset for name, asset in assets_page}, "pagination": { "page": page, "limit": limit, "total": total, "has_more": has_more } } @app.get("/asset/{name}") async def get_asset_by_name_legacy(name: str): """Legacy route - redirect to /assets/{name}.""" return RedirectResponse(url=f"/assets/{name}", status_code=301) @app.get("/assets/{name}") async def get_asset(name: str, request: Request): """Get asset by name. HTML for browsers, JSON for APIs.""" registry = await load_registry() if name not in registry.get("assets", {}): if wants_html(request): content = f'''

Asset Not Found

No asset named "{name}" exists.

← Back to Assets

''' return HTMLResponse(base_html("Asset Not Found", content, get_user_from_cookie(request))) raise HTTPException(404, f"Asset not found: {name}") if wants_html(request): return await ui_asset_detail(name, request) return registry["assets"][name] @app.patch("/assets/{name}") async def update_asset(name: str, req: UpdateAssetRequest, user: User = Depends(get_required_user)): """Update an existing asset's metadata. Creates an Update activity.""" asset = await db.get_asset(name) if not asset: raise HTTPException(404, f"Asset not found: {name}") # Check ownership if asset.get("owner") != user.username: raise HTTPException(403, f"Not authorized to update asset owned by {asset.get('owner')}") # Build updates dict updates = {} if req.description is not None: updates["description"] = req.description if req.tags is not None: updates["tags"] = req.tags if req.metadata is not None: updates["metadata"] = {**asset.get("metadata", {}), **req.metadata} if req.origin is not None: updates["origin"] = req.origin # Update asset in database updated_asset = await db.update_asset(name, updates) # Create Update activity activity = { "activity_id": str(uuid.uuid4()), "activity_type": "Update", "actor_id": f"https://{DOMAIN}/users/{user.username}", "object_data": { "type": updated_asset.get("asset_type", "Object").capitalize(), "name": name, "id": f"https://{DOMAIN}/objects/{updated_asset['content_hash']}", "contentHash": { "algorithm": "sha3-256", "value": updated_asset["content_hash"] }, "attributedTo": f"https://{DOMAIN}/users/{user.username}", "summary": req.description, "tag": req.tags or updated_asset.get("tags", []) }, "published": updated_asset.get("updated_at", datetime.now(timezone.utc).isoformat()) } # Sign activity with the user's keys activity = sign_activity(activity, user.username) # Save activity to database await db.create_activity(activity) return {"asset": updated_asset, "activity": activity} async def _register_asset_impl(req: RegisterRequest, owner: str): """Internal implementation for registering an asset.""" # Check if name exists if await db.asset_exists(req.name): raise HTTPException(400, f"Asset already exists: {req.name}") # Create asset now = datetime.now(timezone.utc).isoformat() asset = { "name": req.name, "content_hash": req.content_hash, "asset_type": req.asset_type, "tags": req.tags, "metadata": req.metadata, "url": req.url, "provenance": req.provenance, "owner": owner, "created_at": now } # Save asset to database created_asset = await db.create_asset(asset) # Create ownership activity object_data = { "type": req.asset_type.capitalize(), "name": req.name, "id": f"https://{DOMAIN}/objects/{req.content_hash}", "contentHash": { "algorithm": "sha3-256", "value": req.content_hash }, "attributedTo": f"https://{DOMAIN}/users/{owner}" } # Include provenance in activity object_data if present if req.provenance: object_data["provenance"] = req.provenance activity = { "activity_id": str(uuid.uuid4()), "activity_type": "Create", "actor_id": f"https://{DOMAIN}/users/{owner}", "object_data": object_data, "published": now } # Sign activity with the owner's keys activity = sign_activity(activity, owner) # Save activity to database await db.create_activity(activity) return {"asset": created_asset, "activity": activity} @app.post("/assets") async def register_asset(req: RegisterRequest, user: User = Depends(get_required_user)): """Register a new asset and create ownership activity. Requires authentication.""" return await _register_asset_impl(req, user.username) @app.post("/assets/record-run") async def record_run(req: RecordRunRequest, user: User = Depends(get_required_user)): """Record an L1 run and register the output. Requires authentication.""" # Fetch run from the specified L1 server l1_url = req.l1_server.rstrip('/') try: resp = requests.get(f"{l1_url}/runs/{req.run_id}") resp.raise_for_status() run = resp.json() except Exception as e: raise HTTPException(400, f"Failed to fetch run from L1 ({l1_url}): {e}") if run.get("status") != "completed": raise HTTPException(400, f"Run not completed: {run.get('status')}") output_hash = run.get("output_hash") if not output_hash: raise HTTPException(400, "Run has no output hash") # Build provenance from run provenance = { "inputs": [{"content_hash": h} for h in run.get("inputs", [])], "recipe": run.get("recipe"), "effect_url": run.get("effect_url"), "effects_commit": run.get("effects_commit"), "l1_server": l1_url, "l1_run_id": req.run_id, "rendered_at": run.get("completed_at"), "infrastructure": run.get("infrastructure") } # Register the output under the authenticated user return await _register_asset_impl(RegisterRequest( name=req.output_name, content_hash=output_hash, asset_type="video", # Could be smarter about this tags=["rendered", "l1"], metadata={"l1_server": l1_url, "l1_run_id": req.run_id}, provenance=provenance ), user.username) @app.post("/assets/publish-cache") async def publish_cache(req: PublishCacheRequest, user: User = Depends(get_required_user)): """ Publish a cache item from L1 with metadata. Requires origin to be set (self or external URL). Creates a new asset and Create activity. """ # Validate origin if not req.origin or "type" not in req.origin: raise HTTPException(400, "Origin is required for publishing (type: 'self' or 'external')") origin_type = req.origin.get("type") if origin_type not in ("self", "external"): raise HTTPException(400, "Origin type must be 'self' or 'external'") if origin_type == "external" and not req.origin.get("url"): raise HTTPException(400, "External origin requires a URL") # Check if asset name already exists if await db.asset_exists(req.asset_name): raise HTTPException(400, f"Asset name already exists: {req.asset_name}") # Create asset now = datetime.now(timezone.utc).isoformat() asset = { "name": req.asset_name, "content_hash": req.content_hash, "asset_type": req.asset_type, "tags": req.tags, "description": req.description, "origin": req.origin, "metadata": req.metadata, "owner": user.username, "created_at": now } # Save asset to database created_asset = await db.create_asset(asset) # Create ownership activity with origin info object_data = { "type": req.asset_type.capitalize(), "name": req.asset_name, "id": f"https://{DOMAIN}/objects/{req.content_hash}", "contentHash": { "algorithm": "sha3-256", "value": req.content_hash }, "attributedTo": f"https://{DOMAIN}/users/{user.username}", "tag": req.tags } if req.description: object_data["summary"] = req.description # Include origin in ActivityPub object if origin_type == "self": object_data["generator"] = { "type": "Application", "name": "Art DAG", "note": "Original content created by the author" } else: object_data["source"] = { "type": "Link", "href": req.origin.get("url"), "name": req.origin.get("note", "External source") } activity = { "activity_id": str(uuid.uuid4()), "activity_type": "Create", "actor_id": f"https://{DOMAIN}/users/{user.username}", "object_data": object_data, "published": now } # Sign activity with the user's keys activity = sign_activity(activity, user.username) # Save activity to database await db.create_activity(activity) return {"asset": created_asset, "activity": activity} # ============ Activities Endpoints ============ @app.get("/activities") async def get_activities(request: Request, page: int = 1, limit: int = 20): """Get activities. HTML for browsers (with infinite scroll), JSON for APIs (with pagination).""" all_activities = await load_activities() total = len(all_activities) # Reverse for newest first all_activities = list(reversed(all_activities)) # Pagination start = (page - 1) * limit end = start + limit activities_page = all_activities[start:end] has_more = end < total if wants_html(request): username = get_user_from_cookie(request) if not activities_page: if page == 1: content = '''

Activities

No activities yet.

''' else: return HTMLResponse("") # Empty for infinite scroll else: rows = "" for i, activity in enumerate(activities_page): activity_index = total - 1 - (start + i) # Original index obj = activity.get("object_data", {}) activity_type = activity.get("activity_type", "") type_color = "bg-green-600" if activity_type == "Create" else "bg-yellow-600" if activity_type == "Update" else "bg-gray-600" actor_id = activity.get("actor_id", "") actor_name = actor_id.split("/")[-1] if actor_id else "unknown" rows += f''' {activity_type} {obj.get("name", "Untitled")} {actor_name} {activity.get("published", "")[:10]} View ''' # For infinite scroll, just return rows if not first page if page > 1: if has_more: rows += f''' Loading more... ''' return HTMLResponse(rows) # First page - full content with header infinite_scroll_trigger = "" if has_more: infinite_scroll_trigger = f''' Loading more... ''' content = f'''

Activities ({total} total)

{rows} {infinite_scroll_trigger}
Type Object Actor Published
''' return HTMLResponse(base_html("Activities", content, username)) # JSON response for APIs return { "activities": activities_page, "pagination": { "page": page, "limit": limit, "total": total, "has_more": has_more } } @app.get("/activities/{activity_index}") async def get_activity_detail(activity_index: int, request: Request): """Get single activity. HTML for browsers, JSON for APIs.""" activities = await load_activities() if activity_index < 0 or activity_index >= len(activities): if wants_html(request): content = '''

Activity Not Found

This activity does not exist.

← Back to Activities

''' return HTMLResponse(base_html("Activity Not Found", content, get_user_from_cookie(request))) raise HTTPException(404, "Activity not found") activity = activities[activity_index] if wants_html(request): # Reuse the UI activity detail logic return await ui_activity_detail(activity_index, request) return activity @app.get("/activity/{activity_index}") async def get_activity_legacy(activity_index: int): """Legacy route - redirect to /activities/{activity_index}.""" return RedirectResponse(url=f"/activities/{activity_index}", status_code=301) @app.get("/objects/{content_hash}") async def get_object(content_hash: str, request: Request): """Get object by content hash. Content negotiation: HTML for browsers, JSON for APIs.""" registry = await load_registry() # Find asset by hash for name, asset in registry.get("assets", {}).items(): if asset.get("content_hash") == content_hash: # Check Accept header for content negotiation accept = request.headers.get("accept", "") wants_html = "text/html" in accept and "application/json" not in accept and "application/activity+json" not in accept if wants_html: # Redirect to detail page for browsers return RedirectResponse(url=f"/assets/{name}", status_code=303) owner = asset.get("owner", "unknown") return JSONResponse( content={ "@context": "https://www.w3.org/ns/activitystreams", "id": f"https://{DOMAIN}/objects/{content_hash}", "type": asset.get("asset_type", "Object").capitalize(), "name": name, "contentHash": { "algorithm": "sha3-256", "value": content_hash }, "attributedTo": f"https://{DOMAIN}/users/{owner}", "published": asset.get("created_at") }, media_type="application/activity+json" ) raise HTTPException(404, f"Object not found: {content_hash}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8200)