Store effects in IPFS instead of local cache

- Upload effects to IPFS, return CID instead of content hash
- Fetch effects from IPFS if not in local cache
- Keep local cache for fast worker access
- Support both :cid (new) and :hash (legacy) in recipes

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-12 07:28:05 +00:00
parent db3faa6a2c
commit c7c7c90909

View File

@@ -2,6 +2,7 @@
Effects routes for L1 server.
Handles effect upload, listing, and metadata.
Effects are stored in IPFS like all other content-addressed data.
"""
import hashlib
@@ -23,6 +24,7 @@ from ..dependencies import (
get_cache_manager,
)
from ..services.auth_service import AuthService
import ipfs_client
router = APIRouter()
logger = logging.getLogger(__name__)
@@ -152,10 +154,10 @@ async def upload_effect(
ctx: UserContext = Depends(require_auth),
):
"""
Upload an effect to the cache.
Upload an effect to IPFS.
Parses PEP 723 metadata and @-tag docstring.
Returns content hash for use in recipes.
Returns IPFS CID for use in recipes.
"""
content = await file.read()
@@ -164,9 +166,6 @@ async def upload_effect(
except UnicodeDecodeError:
raise HTTPException(400, "Effect must be valid UTF-8 text")
# Compute content hash
content_hash = hashlib.sha3_256(content).hexdigest()
# Parse metadata
try:
meta = parse_effect_metadata(source)
@@ -177,17 +176,20 @@ async def upload_effect(
if not meta.get("name"):
meta["name"] = Path(file.filename).stem if file.filename else "unknown"
# Store effect
effects_dir = get_effects_dir()
effect_dir = effects_dir / content_hash
effect_dir.mkdir(parents=True, exist_ok=True)
# Store effect source in IPFS
cid = ipfs_client.add_bytes(content)
if not cid:
raise HTTPException(500, "Failed to store effect in IPFS")
# Write source
# Also keep local cache for fast worker access
effects_dir = get_effects_dir()
effect_dir = effects_dir / cid
effect_dir.mkdir(parents=True, exist_ok=True)
(effect_dir / "effect.py").write_text(source, encoding="utf-8")
# Write metadata
# Store metadata (locally and in IPFS)
full_meta = {
"content_hash": content_hash,
"cid": cid,
"meta": meta,
"uploader": ctx.actor_id,
"uploaded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
@@ -195,10 +197,14 @@ async def upload_effect(
}
(effect_dir / "metadata.json").write_text(json.dumps(full_meta, indent=2))
logger.info(f"Uploaded effect '{meta.get('name')}' hash={content_hash[:16]}... by {ctx.actor_id}")
# Also store metadata in IPFS for discoverability
meta_cid = ipfs_client.add_json(full_meta)
logger.info(f"Uploaded effect '{meta.get('name')}' cid={cid} by {ctx.actor_id}")
return {
"content_hash": content_hash,
"cid": cid,
"metadata_cid": meta_cid,
"name": meta.get("name"),
"version": meta.get("version"),
"temporal": meta.get("temporal", False),
@@ -208,21 +214,35 @@ async def upload_effect(
}
@router.get("/{content_hash}")
@router.get("/{cid}")
async def get_effect(
content_hash: str,
cid: str,
request: Request,
ctx: UserContext = Depends(require_auth),
):
"""Get effect metadata by hash."""
"""Get effect metadata by CID."""
effects_dir = get_effects_dir()
effect_dir = effects_dir / content_hash
effect_dir = effects_dir / cid
metadata_path = effect_dir / "metadata.json"
if not metadata_path.exists():
raise HTTPException(404, f"Effect {content_hash[:16]}... not found")
# Try local cache first
if metadata_path.exists():
meta = json.loads(metadata_path.read_text())
else:
# Fetch from IPFS
source_bytes = ipfs_client.get_bytes(cid)
if not source_bytes:
raise HTTPException(404, f"Effect {cid[:16]}... not found")
meta = json.loads(metadata_path.read_text())
# Cache locally
effect_dir.mkdir(parents=True, exist_ok=True)
source = source_bytes.decode("utf-8")
(effect_dir / "effect.py").write_text(source)
# Parse metadata from source
parsed_meta = parse_effect_metadata(source)
meta = {"cid": cid, "meta": parsed_meta}
(effect_dir / "metadata.json").write_text(json.dumps(meta, indent=2))
if wants_json(request):
return meta
@@ -236,19 +256,30 @@ async def get_effect(
)
@router.get("/{content_hash}/source")
@router.get("/{cid}/source")
async def get_effect_source(
content_hash: str,
cid: str,
ctx: UserContext = Depends(require_auth),
):
"""Get effect source code."""
effects_dir = get_effects_dir()
source_path = effects_dir / content_hash / "effect.py"
source_path = effects_dir / cid / "effect.py"
if not source_path.exists():
raise HTTPException(404, f"Effect {content_hash[:16]}... not found")
# Try local cache first
if source_path.exists():
return PlainTextResponse(source_path.read_text())
return PlainTextResponse(source_path.read_text())
# Fetch from IPFS
source_bytes = ipfs_client.get_bytes(cid)
if not source_bytes:
raise HTTPException(404, f"Effect {cid[:16]}... not found")
# Cache locally
source_path.parent.mkdir(parents=True, exist_ok=True)
source = source_bytes.decode("utf-8")
source_path.write_text(source)
return PlainTextResponse(source)
@router.get("")
@@ -285,17 +316,17 @@ async def list_effects(
)
@router.delete("/{content_hash}")
@router.delete("/{cid}")
async def delete_effect(
content_hash: str,
cid: str,
ctx: UserContext = Depends(require_auth),
):
"""Delete an effect."""
"""Delete an effect from local cache (IPFS content is immutable)."""
effects_dir = get_effects_dir()
effect_dir = effects_dir / content_hash
effect_dir = effects_dir / cid
if not effect_dir.exists():
raise HTTPException(404, f"Effect {content_hash[:16]}... not found")
raise HTTPException(404, f"Effect {cid[:16]}... not found in local cache")
# Check ownership
metadata_path = effect_dir / "metadata.json"
@@ -307,5 +338,8 @@ async def delete_effect(
import shutil
shutil.rmtree(effect_dir)
logger.info(f"Deleted effect {content_hash[:16]}... by {ctx.actor_id}")
return {"deleted": True}
# Unpin from IPFS (content remains available if pinned elsewhere)
ipfs_client.unpin(cid)
logger.info(f"Deleted effect {cid[:16]}... by {ctx.actor_id}")
return {"deleted": True, "note": "Unpinned from local IPFS; content may still exist on other nodes"}