feat: add zip file extraction support with error handling for invalid archives

devplacepy/services/devii/actions/catalog.py

@@ -280,6 +280,30 @@ ACTIONS: tuple[Action, ...] = (
             body("path", "Relative path to delete.", required=True),
         ),
     ),
+    Action(
+        name="zip_project",
+        method="POST",
+        path="/projects/{project_slug}/zip",
+        summary="Build a downloadable zip archive of a whole project",
+        description=(
+            "Queues a background zip job and returns {uid, status_url}. Poll the status_url with "
+            "zip_status until status is 'done', then give the user the download_url."
+        ),
+        params=(path("project_slug", "Project slug or uid."),),
+        requires_auth=False,
+    ),
+    Action(
+        name="zip_status",
+        method="GET",
+        path="/zips/{uid}",
+        summary="Check a zip job and obtain its download link once finished",
+        description=(
+            "Returns the job status and stats. When status is 'done', download_url points at the "
+            "ready archive; while 'pending' or 'running', poll again shortly."
+        ),
+        params=(path("uid", "Zip job uid returned by zip_project."),),
+        requires_auth=False,
+    ),
     Action(
         name="search_users",
         method="GET",

devplacepy/services/jobs/__init__.py

@@ -0,0 +1 @@
+# retoor <retoor@molodetz.nl>

devplacepy/services/jobs/base.py

@@ -0,0 +1,246 @@
+# retoor <retoor@molodetz.nl>
+
+import asyncio
+import json
+import logging
+from datetime import datetime, timedelta, timezone
+
+from devplacepy.database import get_table, get_int_setting
+from devplacepy.services.base import BaseService, ConfigField
+from devplacepy.services.jobs import queue
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_RETENTION_SECONDS = 7 * 24 * 60 * 60
+DEFAULT_MAX_CONCURRENT = 2
+DEFAULT_JOB_TIMEOUT_SECONDS = 600
+
+
+class JobService(BaseService):
+    kind = ""
+    min_interval = 1
+    max_retries = 3
+
+    def __init__(self, name: str, interval_seconds: int = 2):
+        super().__init__(name=name, interval_seconds=interval_seconds)
+        self.kind = self.kind or name
+        self.retention_key = f"{name}_retention_seconds"
+        self.concurrency_key = f"{name}_max_concurrent"
+        self.timeout_key = f"{name}_job_timeout_seconds"
+        self.config_fields = [
+            ConfigField(self.retention_key, "Artifact retention (seconds)", type="int",
+                        default=DEFAULT_RETENTION_SECONDS, minimum=60,
+                        help="Finished artifacts are deleted once unused for this long.",
+                        group="Jobs"),
+            ConfigField(self.concurrency_key, "Max concurrent jobs", type="int",
+                        default=DEFAULT_MAX_CONCURRENT, minimum=1, maximum=16,
+                        help="Number of jobs processed in parallel.",
+                        group="Jobs"),
+            ConfigField(self.timeout_key, "Job timeout (seconds)", type="int",
+                        default=DEFAULT_JOB_TIMEOUT_SECONDS, minimum=10,
+                        help="A running job with no live task after this long is retried.",
+                        group="Jobs"),
+        ]
+        self._inflight: dict = {}
+
+    def retention_seconds(self) -> int:
+        return max(60, get_int_setting(self.retention_key, DEFAULT_RETENTION_SECONDS))
+
+    def max_concurrent(self) -> int:
+        return max(1, get_int_setting(self.concurrency_key, DEFAULT_MAX_CONCURRENT))
+
+    def job_timeout_seconds(self) -> int:
+        return max(10, get_int_setting(self.timeout_key, DEFAULT_JOB_TIMEOUT_SECONDS))
+
+    async def process(self, job: dict) -> dict:
+        raise NotImplementedError
+
+    def cleanup(self, job: dict) -> None:
+        pass
+
+    async def on_enable(self) -> None:
+        self._recover_orphans(min_age_seconds=0)
+
+    async def on_disable(self) -> None:
+        for entry in list(self._inflight.values()):
+            entry["task"].cancel()
+        self._inflight.clear()
+
+    async def run_once(self) -> None:
+        self._reap()
+        self._recover_orphans(min_age_seconds=self.job_timeout_seconds())
+        self._refill()
+        self._sweep_expired()
+
+    def _reap(self) -> None:
+        for uid in list(self._inflight):
+            entry = self._inflight[uid]
+            task = entry["task"]
+            if not task.done():
+                continue
+            del self._inflight[uid]
+            duration_ms = int((datetime.now(timezone.utc) - entry["started"]).total_seconds() * 1000)
+            exc = task.exception() if not task.cancelled() else asyncio.CancelledError()
+            if exc is not None:
+                self._finish_failed(uid, str(exc) or exc.__class__.__name__, duration_ms)
+            else:
+                self._finish_done(uid, task.result() or {}, duration_ms)
+
+    def _finish_done(self, uid: str, result_data: dict, duration_ms: int) -> None:
+        now = datetime.now(timezone.utc)
+        get_table("jobs").update({
+            "uid": uid,
+            "status": queue.DONE,
+            "result": json.dumps(result_data),
+            "error": "",
+            "completed_at": now.isoformat(),
+            "updated_at": now.isoformat(),
+            "duration_ms": duration_ms,
+            "last_accessed_at": now.isoformat(),
+            "expires_at": (now + timedelta(seconds=self.retention_seconds())).isoformat(),
+            "bytes_in": int(result_data.get("bytes_in", 0)),
+            "bytes_out": int(result_data.get("bytes_out", 0)),
+            "item_count": int(result_data.get("item_count", 0)),
+        }, ["uid"])
+        self.log(f"Job {uid} done in {duration_ms}ms")
+
+    def _finish_failed(self, uid: str, error: str, duration_ms: int) -> None:
+        now = datetime.now(timezone.utc)
+        get_table("jobs").update({
+            "uid": uid,
+            "status": queue.FAILED,
+            "error": error[:2000],
+            "completed_at": now.isoformat(),
+            "updated_at": now.isoformat(),
+            "duration_ms": duration_ms,
+        }, ["uid"])
+        self.log(f"Job {uid} failed: {error}")
+
+    def _refill(self) -> None:
+        capacity = self.max_concurrent() - len(self._inflight)
+        if capacity <= 0:
+            return
+        table = get_table("jobs")
+        pending = list(table.find(kind=self.kind, status=queue.PENDING, order_by=["uid"], _limit=capacity))
+        for row in pending:
+            uid = row["uid"]
+            if uid in self._inflight:
+                continue
+            now = datetime.now(timezone.utc)
+            table.update({
+                "uid": uid,
+                "status": queue.RUNNING,
+                "started_at": now.isoformat(),
+                "updated_at": now.isoformat(),
+                "error": "",
+            }, ["uid"])
+            job = queue.get_job(uid)
+            self._inflight[uid] = {"task": asyncio.create_task(self._run_job(job)), "started": now}
+            self.log(f"Job {uid} started")
+
+    async def _run_job(self, job: dict) -> dict:
+        return await self.process(job)
+
+    def _recover_orphans(self, min_age_seconds: int) -> None:
+        table = get_table("jobs")
+        now = datetime.now(timezone.utc)
+        for row in table.find(kind=self.kind, status=queue.RUNNING):
+            uid = row["uid"]
+            if uid in self._inflight:
+                continue
+            if min_age_seconds > 0 and not self._older_than(row.get("started_at"), now, min_age_seconds):
+                continue
+            retry_count = int(row.get("retry_count") or 0)
+            if retry_count >= self.max_retries:
+                self._finish_failed(uid, "exceeded retry limit after orphan recovery", 0)
+                continue
+            table.update({
+                "uid": uid,
+                "status": queue.PENDING,
+                "retry_count": retry_count + 1,
+                "started_at": "",
+                "updated_at": now.isoformat(),
+            }, ["uid"])
+            self.log(f"Recovered orphaned job {uid} (retry {retry_count + 1})")
+
+    def _sweep_expired(self) -> None:
+        table = get_table("jobs")
+        now = datetime.now(timezone.utc)
+        for row in table.find(kind=self.kind, status=queue.DONE):
+            expires_at = row.get("expires_at")
+            if not expires_at or not self._is_past(expires_at, now):
+                continue
+            job = queue.get_job(row["uid"])
+            try:
+                self.cleanup(job)
+            except Exception as exc:
+                self.log(f"Cleanup failed for {row['uid']}: {exc}")
+            table.delete(uid=row["uid"])
+            self.log(f"Pruned expired job {row['uid']}")
+
+    def _older_than(self, stamp: str, now: datetime, seconds: int) -> bool:
+        moment = self._parse(stamp)
+        if moment is None:
+            return True
+        return (now - moment).total_seconds() >= seconds
+
+    def _is_past(self, stamp: str, now: datetime) -> bool:
+        moment = self._parse(stamp)
+        return moment is not None and moment < now
+
+    def _parse(self, stamp: str):
+        if not stamp:
+            return None
+        try:
+            return datetime.fromisoformat(stamp)
+        except (ValueError, TypeError):
+            return None
+
+    def collect_metrics(self) -> dict:
+        rows = queue.list_jobs(kind=self.kind)
+        by_status = {queue.PENDING: 0, queue.RUNNING: 0, queue.DONE: 0, queue.FAILED: 0}
+        bytes_in = bytes_out = items = 0
+        durations = []
+        for row in rows:
+            by_status[row.get("status", "")] = by_status.get(row.get("status", ""), 0) + 1
+            bytes_in += int(row.get("bytes_in") or 0)
+            bytes_out += int(row.get("bytes_out") or 0)
+            items += int(row.get("item_count") or 0)
+            if row.get("status") == queue.DONE and row.get("duration_ms"):
+                durations.append(int(row["duration_ms"]))
+        avg_ms = int(sum(durations) / len(durations)) if durations else 0
+        stats = [
+            {"label": "Total jobs", "value": len(rows)},
+            {"label": "Pending", "value": by_status[queue.PENDING]},
+            {"label": "Running", "value": by_status[queue.RUNNING]},
+            {"label": "Done", "value": by_status[queue.DONE]},
+            {"label": "Failed", "value": by_status[queue.FAILED]},
+            {"label": "Items zipped", "value": items},
+            {"label": "Bytes in", "value": _human_bytes(bytes_in)},
+            {"label": "Bytes out", "value": _human_bytes(bytes_out)},
+            {"label": "Avg duration", "value": f"{avg_ms} ms"},
+        ]
+        recent = sorted(rows, key=lambda r: r.get("uid", ""), reverse=True)[:10]
+        table = {
+            "columns": ["Job", "Status", "Name", "Out", "Items"],
+            "rows": [
+                [
+                    row.get("uid", "")[:8],
+                    row.get("status", ""),
+                    (row.get("preferred_name") or "")[:32],
+                    _human_bytes(int(row.get("bytes_out") or 0)),
+                    int(row.get("item_count") or 0),
+                ]
+                for row in recent
+            ],
+        }
+        return {"stats": stats, "table": table}
+
+
+def _human_bytes(size: int) -> str:
+    value = float(size)
+    for unit in ("B", "KB", "MB", "GB", "TB"):
+        if value < 1024 or unit == "TB":
+            return f"{value:.0f} {unit}" if unit == "B" else f"{value:.1f} {unit}"
+        value /= 1024
+    return f"{value:.1f} TB"

devplacepy/services/jobs/queue.py

@@ -0,0 +1,103 @@
+# retoor <retoor@molodetz.nl>
+
+import json
+import logging
+from datetime import datetime, timedelta, timezone
+
+from devplacepy.database import db, get_table
+from devplacepy.utils import generate_uid
+
+logger = logging.getLogger(__name__)
+
+PENDING = "pending"
+RUNNING = "running"
+DONE = "done"
+FAILED = "failed"
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _table():
+    return get_table("jobs")
+
+
+def _decode(field: str) -> dict:
+    if not field:
+        return {}
+    try:
+        return json.loads(field)
+    except (ValueError, TypeError):
+        return {}
+
+
+def _hydrate(row: dict) -> dict:
+    row = dict(row)
+    row["payload"] = _decode(row.get("payload"))
+    row["result"] = _decode(row.get("result"))
+    return row
+
+
+def enqueue(kind: str, payload: dict, owner_kind: str, owner_id: str, preferred_name: str = "") -> str:
+    uid = generate_uid()
+    now = _now()
+    _table().insert({
+        "uid": uid,
+        "kind": kind,
+        "status": PENDING,
+        "owner_kind": owner_kind,
+        "owner_id": owner_id,
+        "preferred_name": preferred_name,
+        "payload": json.dumps(payload),
+        "result": "",
+        "error": "",
+        "retry_count": 0,
+        "created_at": now,
+        "started_at": "",
+        "completed_at": "",
+        "updated_at": now,
+        "duration_ms": 0,
+        "last_accessed_at": "",
+        "expires_at": "",
+        "bytes_in": 0,
+        "bytes_out": 0,
+        "item_count": 0,
+    })
+    logger.info("Enqueued %s job %s for %s/%s", kind, uid, owner_kind, owner_id)
+    return uid
+
+
+def get_job(uid: str) -> dict | None:
+    if "jobs" not in db.tables:
+        return None
+    row = _table().find_one(uid=uid)
+    return _hydrate(row) if row else None
+
+
+def touch_job(uid: str, extend_seconds: int) -> None:
+    if "jobs" not in db.tables:
+        return
+    row = _table().find_one(uid=uid)
+    if not row:
+        return
+    now = datetime.now(timezone.utc)
+    _table().update({
+        "uid": uid,
+        "last_accessed_at": now.isoformat(),
+        "expires_at": (now + timedelta(seconds=extend_seconds)).isoformat(),
+        "updated_at": now.isoformat(),
+    }, ["uid"])
+
+
+def list_jobs(kind: str = None, status: str = None, owner: tuple = None) -> list[dict]:
+    if "jobs" not in db.tables:
+        return []
+    filters = {}
+    if kind:
+        filters["kind"] = kind
+    if status:
+        filters["status"] = status
+    if owner:
+        filters["owner_kind"], filters["owner_id"] = owner
+    return [_hydrate(row) for row in _table().find(**filters)]

devplacepy/services/jobs/zip_service.py

@@ -0,0 +1,96 @@
+# retoor <retoor@molodetz.nl>
+
+import asyncio
+import json
+import logging
+import os
+import shutil
+import sys
+from pathlib import Path
+
+from devplacepy.config import BASE_DIR, STATIC_DIR
+from devplacepy import project_files
+from devplacepy.services.jobs.base import JobService
+from devplacepy.utils import generate_uid, slugify
+
+logger = logging.getLogger(__name__)
+
+ZIPS_DIR = STATIC_DIR / "uploads" / "zips"
+STAGING_DIR = STATIC_DIR / "uploads" / "zip_staging"
+WORKER_MODULE = "devplacepy.services.jobs.zip_worker"
+
+
+def _shard(uid: str) -> str:
+    return f"{uid[:2]}/{uid[2:4]}"
+
+
+class ZipService(JobService):
+    kind = "zip"
+    title = "Zip"
+    description = (
+        "Builds downloadable zip archives in a subprocess off the request path, tracks each "
+        "job with full statistics, and deletes archives once they go unused for the retention "
+        "window."
+    )
+
+    def __init__(self):
+        super().__init__(name="zip", interval_seconds=2)
+
+    async def process(self, job: dict) -> dict:
+        source = job["payload"].get("source", {})
+        uid = job["uid"]
+        staging = STAGING_DIR / uid
+        try:
+            item_count = await asyncio.to_thread(self._materialize, source, staging)
+            tmp_dir = ZIPS_DIR / _shard(uid)
+            tmp_dir.mkdir(parents=True, exist_ok=True)
+            tmp_zip = tmp_dir / f"{generate_uid()}.zip"
+            stats = await self._run_worker(staging, tmp_zip)
+            final_name = self._final_name(job.get("preferred_name", ""), stats["crc32"])
+            final_path = tmp_dir / final_name
+            os.replace(tmp_zip, final_path)
+        finally:
+            await asyncio.to_thread(shutil.rmtree, staging, ignore_errors=True)
+
+        return {
+            "download_url": f"/zips/{uid}/download",
+            "local_path": str(final_path),
+            "final_name": final_name,
+            "crc32": f"{stats['crc32']:08x}",
+            "file_count": stats["file_count"],
+            "dir_count": stats["dir_count"],
+            "bytes_in": stats["bytes_in"],
+            "bytes_out": stats["bytes_out"],
+            "item_count": item_count,
+        }
+
+    def _materialize(self, source: dict, staging: Path) -> int:
+        if source.get("type") == "project_tree":
+            return project_files.export_to_dir(source["project_uid"], source.get("path", ""), staging)
+        raise ValueError(f"unsupported zip source: {source.get('type')}")
+
+    async def _run_worker(self, source_dir: Path, output_path: Path) -> dict:
+        proc = await asyncio.create_subprocess_exec(
+            sys.executable, "-m", WORKER_MODULE, str(source_dir), str(output_path),
+            cwd=str(BASE_DIR),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        out, err = await proc.communicate()
+        if proc.returncode != 0:
+            raise RuntimeError(f"zip worker exited {proc.returncode}: {err.decode('utf-8', 'replace')[:500]}")
+        return json.loads(out.decode("utf-8"))
+
+    def _final_name(self, preferred_name: str, crc32: int) -> str:
+        name = preferred_name or "download"
+        if name.lower().endswith(".zip"):
+            name = name[:-4]
+        slug = slugify(name) or "download"
+        return f"{crc32:08x}.{slug}.zip"
+
+    def cleanup(self, job: dict) -> None:
+        result = job.get("result", {})
+        local_path = result.get("local_path")
+        if local_path:
+            Path(local_path).unlink(missing_ok=True)
+        shutil.rmtree(STAGING_DIR / job["uid"], ignore_errors=True)

devplacepy/services/jobs/zip_worker.py

@@ -0,0 +1,65 @@
+# retoor <retoor@molodetz.nl>
+
+import json
+import os
+import shutil
+import sys
+import zipfile
+import zlib
+
+FIXED_DATE = (1980, 1, 1, 0, 0, 0)
+
+
+def _build(source_dir: str, output_path: str) -> dict:
+    bytes_in = 0
+    file_count = 0
+    dir_count = 0
+    with zipfile.ZipFile(output_path, "w", zipfile.ZIP_DEFLATED) as archive:
+        for root, dirs, files in os.walk(source_dir):
+            dirs.sort()
+            for name in dirs:
+                dir_count += 1
+                arcname = os.path.relpath(os.path.join(root, name), source_dir) + "/"
+                info = zipfile.ZipInfo(arcname, date_time=FIXED_DATE)
+                info.external_attr = (0o040755 << 16) | 0x10
+                archive.writestr(info, b"")
+            for name in sorted(files):
+                full = os.path.join(root, name)
+                arcname = os.path.relpath(full, source_dir)
+                bytes_in += os.path.getsize(full)
+                info = zipfile.ZipInfo(arcname, date_time=FIXED_DATE)
+                info.compress_type = zipfile.ZIP_DEFLATED
+                info.external_attr = 0o644 << 16
+                with open(full, "rb") as source, archive.open(info, "w") as target:
+                    shutil.copyfileobj(source, target)
+                file_count += 1
+
+    crc = 0
+    with open(output_path, "rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            crc = zlib.crc32(chunk, crc)
+
+    return {
+        "bytes_in": bytes_in,
+        "bytes_out": os.path.getsize(output_path),
+        "file_count": file_count,
+        "dir_count": dir_count,
+        "crc32": crc & 0xFFFFFFFF,
+    }
+
+
+def main(argv: list) -> int:
+    if len(argv) != 3:
+        sys.stderr.write("usage: zip_worker <source_dir> <output_zip>\n")
+        return 2
+    source_dir, output_path = argv[1], argv[2]
+    if not os.path.isdir(source_dir):
+        sys.stderr.write(f"source directory not found: {source_dir}\n")
+        return 3
+    stats = _build(source_dir, output_path)
+    sys.stdout.write(json.dumps(stats))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv))