diff --git a/coulomb_social/apps/spaces/content.py b/coulomb_social/apps/spaces/content.py index d0c2f40..b463f4f 100644 --- a/coulomb_social/apps/spaces/content.py +++ b/coulomb_social/apps/spaces/content.py @@ -88,6 +88,13 @@ def page_path_for(space: Space, page_slug: str) -> str: def render_markdown(text: str) -> str: + """Render the supported Markdown subset. + + Extensions are fenced code, tables, newline-to-break and sane lists. + Raw HTML, script, event handlers and non-http(s)/mailto URLs are removed. + Relative asset URLs are kept so copied page images still resolve. This is + not acceptance of a private corpus or of arbitrary Bubble markup. + """ raw_html = md_lib.markdown( text, extensions=["fenced_code", "tables", "nl2br", "sane_lists"], diff --git a/coulomb_social/apps/spaces/pageops.py b/coulomb_social/apps/spaces/pageops.py index 7e1ca6c..dd01854 100644 --- a/coulomb_social/apps/spaces/pageops.py +++ b/coulomb_social/apps/spaces/pageops.py @@ -21,8 +21,10 @@ import re import shutil import tempfile import threading +import uuid from contextlib import contextmanager from dataclasses import dataclass, field +from datetime import UTC, datetime from functools import wraps from pathlib import Path from urllib.parse import quote, unquote, urlsplit @@ -42,7 +44,7 @@ def content_write_lock(): """Serialize cooperating app/import workers on one mounted content instance.""" with _WRITER_LOCK: if getattr(_WRITER_STATE, "active", False): - yield + yield False return root = content_root() root.mkdir(parents=True, exist_ok=True) @@ -50,16 +52,20 @@ def content_write_lock(): try: fcntl.flock(fd, fcntl.LOCK_EX) _WRITER_STATE.active = True - yield + try: + yield True + finally: + _WRITER_STATE.active = False finally: - _WRITER_STATE.active = False os.close(fd) def content_writer(function): @wraps(function) def locked(*args, **kwargs): - with content_write_lock(): + with content_write_lock() as outermost: + if outermost and function.__name__ != "recover_interrupted_content": + _recover_interrupted_content() return function(*args, **kwargs) return locked @@ -251,14 +257,8 @@ def page_aliases(space_slug): raise PageOpsError("Page-address history needs operator review") from exc -@content_writer -def remember_page_address(source_space, source_page, dest_space, dest_page): - source_page = normalize_slug(source_page) - dest_page = normalize_slug(dest_page) - space_dir(dest_space) - aliases = page_aliases(source_space) - aliases[source_page] = [dest_space, dest_page] - directory = space_dir(source_space) +def _store_aliases(space_slug, aliases): + directory = space_dir(space_slug) temporary = None try: with tempfile.NamedTemporaryFile(dir=directory, delete=False) as stream: @@ -273,6 +273,295 @@ def remember_page_address(source_space, source_page, dest_space, dest_page): temporary.unlink(missing_ok=True) +@content_writer +def remember_page_address(source_space, source_page, dest_space, dest_page): + source_page = normalize_slug(source_page) + dest_page = normalize_slug(dest_page) + space_dir(dest_space) + aliases = page_aliases(source_space) + aliases[source_page] = [dest_space, dest_page] + _store_aliases(source_space, aliases) + + +def _recovery_dir(): + path = content_root() / ".recovery" + if path.is_symlink(): + raise PageOpsError("Content paths must not be symlinks") + path.mkdir(parents=True, exist_ok=True) + return path + + +def _write_record(record): + directory = _recovery_dir() + destination = directory / f"{record['id']}.json" + if destination.is_symlink(): + raise PageOpsError("Content paths must not be symlinks") + temporary = None + try: + with tempfile.NamedTemporaryFile(dir=directory, delete=False) as stream: + temporary = Path(stream.name) + payload = (json.dumps(record, sort_keys=True) + "\n").encode("utf-8") + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, destination) + _fsync_directory(directory) + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + return record + + +def _read_record(record_id): + path = _recovery_dir() / f"{record_id}.json" + if path.is_symlink() or not path.is_file(): + return None + try: + record = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError, TypeError): + return "held" + if not isinstance(record, dict) or record.get("id") != record_id: + return "held" + return record + + +def _clear_record(record): + path = _recovery_dir() / f"{record['id']}.json" + if path.is_symlink(): + raise PageOpsError("Content paths must not be symlinks") + path.unlink(missing_ok=True) + _fsync_directory(path.parent) + + +def _start_record(op, source_space, source_page, dest_space, dest_page, source_revision): + return _write_record({ + "id": uuid.uuid4().hex, + "version": 1, + "op": op, + "source_space": source_space, + "source_page": source_page, + "dest_space": dest_space, + "dest_page": dest_page, + "source_revision": source_revision, + "dest_revision": "", + "assets": [], + "phase": "started", + }) + + +def _note_asset(record, name): + if not isinstance(name, str) or not name or "/" in name or name in (".", ".."): + raise PageOpsError("Recovery asset name needs operator review") + if name not in record["assets"]: + record["assets"].append(name) + _write_record(record) + + +def _file_revision(path): + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _discard_recorded_assets(space_slug, names): + if not names: + return + root = assets_dir(space_slug) + for name in names: + if not isinstance(name, str) or not name or "/" in name or name in (".", ".."): + raise PageOpsError("Recovery asset name needs operator review") + path = root / name + if path.is_symlink(): + raise PageOpsError("Content paths must not be symlinks") + if path.is_file(): + path.unlink() + _fsync_directory(root) + + +def _discard_unconfirmed_page(space_slug, page_slug): + path = page_file(space_slug, page_slug) + if path.is_symlink(): + raise PageOpsError("Content paths must not be symlinks") + if not path.is_file(): + return + _remember(space_slug, page_slug, path.read_bytes()) + path.unlink() + _fsync_directory(path.parent) + + +def _drop_matching_alias(record): + aliases = page_aliases(record["source_space"]) + expected = [record["dest_space"], record["dest_page"]] + if aliases.get(record["source_page"]) != expected: + return + aliases.pop(record["source_page"]) + _store_aliases(record["source_space"], aliases) + + +def _record_is_actionable(record): + if not isinstance(record, dict) or record.get("version") != 1: + return False + if record.get("op") not in {"copy", "transfer"}: + return False + if record.get("phase") not in {"started", "published", "aliased"}: + return False + try: + for key in ("source_space", "dest_space"): + if not _SLUG_RE.fullmatch(record[key]): + return False + for key in ("source_page", "dest_page"): + normalize_slug(record[key]) + except (KeyError, TypeError, PageOpsError): + return False + if record["dest_page"] == "index": + return False + if record["op"] == "transfer" and record["source_page"] == "index": + return False + if not isinstance(record.get("assets"), list): + return False + return isinstance(record.get("source_revision"), str) + + +def _finish_transfer_record(record, source_path, dest_confirmed): + """Complete a published move, or keep a source that changed after it started.""" + source_same = source_path.is_file() and _file_revision(source_path) == record["source_revision"] + if source_same and dest_confirmed: + if record["phase"] != "aliased": + remember_page_address( + record["source_space"], record["source_page"], + record["dest_space"], record["dest_page"], + ) + if _file_revision(source_path) != record["source_revision"]: + return "held" + delete_page(record["source_space"], record["source_page"]) + _clear_record(record) + return "cleared" + if source_path.is_file() and not source_same: + _drop_matching_alias(record) + _clear_record(record) + return "cleared" + if dest_confirmed or page_file(record["dest_space"], record["dest_page"]).is_file(): + _clear_record(record) + return "cleared" + _discard_recorded_assets(record["dest_space"], record["assets"]) + _clear_record(record) + return "cleared" + + +def _recover_record(record): + if record == "held" or not _record_is_actionable(record): + logger.error("content recovery left a journal for operator review") + return "held" + source_path = page_file(record["source_space"], record["source_page"]) + dest_path = page_file(record["dest_space"], record["dest_page"]) + dest_confirmed = bool( + record.get("dest_revision") + and dest_path.is_file() + and _file_revision(dest_path) == record["dest_revision"] + ) + source_exists = source_path.is_file() + if record["op"] == "copy": + if dest_confirmed or (not source_exists and dest_path.is_file()): + _clear_record(record) + return "cleared" + if source_exists: + _discard_unconfirmed_page(record["dest_space"], record["dest_page"]) + _discard_recorded_assets(record["dest_space"], record["assets"]) + _clear_record(record) + return "cleared" + if dest_confirmed: + return _finish_transfer_record(record, source_path, dest_confirmed) + if record["phase"] == "aliased" and source_exists: + _drop_matching_alias(record) + if source_exists: + _discard_unconfirmed_page(record["dest_space"], record["dest_page"]) + _discard_recorded_assets(record["dest_space"], record["assets"]) + _clear_record(record) + return "cleared" + if dest_path.is_file(): + _clear_record(record) + return "cleared" + _discard_recorded_assets(record["dest_space"], record["assets"]) + _clear_record(record) + return "cleared" + + +def _recover_interrupted_content(): + directory = content_root() / ".recovery" + if not directory.exists(): + return {"cleared": 0, "held": 0} + if directory.is_symlink(): + raise PageOpsError("Content paths must not be symlinks") + cleared = held = 0 + for path in sorted(directory.glob("*.json")): + if path.is_symlink(): + held += 1 + logger.error("content recovery left a journal for operator review") + continue + try: + record = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError, TypeError): + held += 1 + logger.error("content recovery left a journal for operator review") + continue + if _recover_record(record) == "held": + held += 1 + else: + cleared += 1 + return {"cleared": cleared, "held": held} + + +@content_writer +def recover_interrupted_content(): + """Finish or roll back a copy/move whose process stopped between files. + + A confirmed destination is kept. The source is removed only when its bytes + still match the journal. Address aliases are written or removed only for + that journal's source page, and they do not grant destination access. + """ + return _recover_interrupted_content() + + +@content_writer +def purge_revisions(*, older_than): + """Delete expired inactive revisions. Never runs as part of a page save. + + The live page bytes are kept. Every stored revision of a page that no + longer has a file is kept, because that directory is its recovery copy. + A revision whose bytes do not match its content-address name stops the + purge before anything is deleted. + """ + if older_than.tzinfo is None: + raise PageOpsError("Revision purge cutoff must include a timezone") + root = content_root() / ".revisions" + if not root.exists(): + return 0 + if root.is_symlink(): + raise PageOpsError("Content paths must not be symlinks") + victims = [] + for path in sorted(root.rglob("*.md")): + if path.is_symlink() or any(parent.is_symlink() for parent in path.parents): + raise PageOpsError("Content paths must not be symlinks") + relative = path.relative_to(root) + if len(relative.parts) != 3: + raise PageOpsError("Stored revision integrity check failed") + data = path.read_bytes() + digest = hashlib.sha256(data).hexdigest() + if path.stem != digest: + raise PageOpsError("Stored revision integrity check failed") + space_slug, page_slug, _name = relative.parts + live = page_file(space_slug, page_slug) + if live.is_file() and _file_revision(live) == digest: + continue + if not live.is_file(): + continue + modified = datetime.fromtimestamp(path.stat().st_mtime, tz=UTC) + if modified < older_than: + victims.append(path) + for path in victims: + path.unlink() + _fsync_directory(path.parent) + return len(victims) + + @content_writer def write_page( space_slug: str, @@ -371,7 +660,7 @@ _BODY_ASSET = re.compile( ) -def _copy_assets(source_space, dest_space, dest_page, visual, body): +def _copy_assets(source_space, dest_space, dest_page, visual, body, on_created=None): """Copy local Visual and Markdown link/image destinations as one staged set.""" from .assets import asset_parts, open_local_asset @@ -408,6 +697,8 @@ def _copy_assets(source_space, dest_space, dest_page, visual, body): stream.flush() os.fsync(stream.fileno()) copied[path] = f"assets/{quote(dest_name)}" + if on_created is not None: + on_created(dest_name) suffix = (f"?{parsed.query}" if parsed.query else "") suffix += f"#{parsed.fragment}" if parsed.fragment else "" return copied[path] + suffix @@ -437,6 +728,8 @@ def copy_page( dest_page: str, *, title: str | None = None, + journal: bool = True, + on_asset=None, ) -> PageDoc: src = read_page(source_space, source_page) if not src.exists: @@ -446,20 +739,42 @@ def copy_page( or dest_page in page_aliases(dest_space)): raise PageOpsError("Destination page already exists") ensure_space_tree(dest_space) - visual, body, created = _copy_assets( - source_space, dest_space, dest_page, src.visual, src.body, - ) - new_title = title if title is not None else (src.title or dest_page) - if title is None and new_title == src.title and source_space == dest_space: - new_title = f"{src.title} (copy)" if src.title else f"{dest_page}" + source_revision = _file_revision(Path(src.path)) + record = _start_record( + "copy", source_space, src.slug, dest_space, dest_page, source_revision, + ) if journal else None + created = [] + + def note_asset(name): + if record is not None: + _note_asset(record, name) + if on_asset is not None: + on_asset(name) + try: - return write_page( + visual, body, created = _copy_assets( + source_space, dest_space, dest_page, src.visual, src.body, note_asset, + ) + new_title = title if title is not None else (src.title or dest_page) + if title is None and new_title == src.title and source_space == dest_space: + new_title = f"{src.title} (copy)" if src.title else f"{dest_page}" + doc = write_page( dest_space, dest_page, title=new_title, abstractor=src.abstractor, visual=visual, body=body, expected_revision="", extra_meta={k: v for k, v in src.meta.items() if k not in {"title", "abstractor", "visual", "space"}}, ) + if record is not None: + record["dest_revision"] = doc.revision + record["phase"] = "published" + _write_record(record) + _clear_record(record) + return doc except BaseException: + if record is not None: + stored = _read_record(record["id"]) + if stored is not None: + _recover_record(stored) if not page_file(dest_space, dest_page).exists(): for path in created: path.unlink(missing_ok=True) @@ -478,10 +793,39 @@ def transfer_page( if source_page == "index": raise PageOpsError("Cannot transfer the space index page") src = read_page(source_space, source_page) - copy_page(source_space, source_page, dest_space, dest_page, title=src.title) - remember_page_address(source_space, source_page, dest_space, dest_page) - delete_page(source_space, source_page) - return read_page(dest_space, dest_page) + if not src.exists: + raise PageOpsError("Source page not found") + # The journal is authority to remove an unconfirmed destination. Refuse an + # existing page before writing that journal. + if (page_file(dest_space, dest_page).is_file() + or dest_page in page_aliases(dest_space)): + raise PageOpsError("Destination page already exists") + record = _start_record( + "transfer", source_space, source_page, dest_space, dest_page, + _file_revision(Path(src.path)), + ) + try: + doc = copy_page( + source_space, source_page, dest_space, dest_page, title=src.title, + journal=False, on_asset=lambda name: _note_asset(record, name), + ) + record["dest_revision"] = doc.revision + record["phase"] = "published" + _write_record(record) + remember_page_address(source_space, source_page, dest_space, dest_page) + record["phase"] = "aliased" + _write_record(record) + source_path = page_file(source_space, source_page) + if _file_revision(source_path) != record["source_revision"]: + raise PageOpsError("Source changed before the move finished") + delete_page(source_space, source_page) + _clear_record(record) + return read_page(dest_space, dest_page) + except BaseException: + stored = _read_record(record["id"]) + if stored is not None: + _recover_record(stored) + raise @content_writer diff --git a/docs/session-pickup.md b/docs/session-pickup.md index 5c6ad40..0acecbc 100644 --- a/docs/session-pickup.md +++ b/docs/session-pickup.md @@ -6,9 +6,10 @@ superseded. Continue work in the registered residual owners: WP-0012 recovery/ service readiness, WP-0013 sharing/content/portability, WP-0014 further cohorts/ Bubble off-ramp, and WP-0015 communication operations/flow acceptance. -2026-10-11: WP-0013 is active. T01 has the global slug constraint and migration -refusal in the checkout; the live database is not migrated. T02 and T03 remain -todo. WP-0012-T01 and WP-0015-T01/T03 still wait on their named owners. +2026-10-11: WP-0013 is active. T01 has the global slug constraint in the checkout; +the live database is not migrated. T02 has revision purge and copy/move crash +recovery in the checkout; private corpus review and any live purge remain open. +T03 remains todo. WP-0012-T01 and WP-0015-T01/T03 still wait on their named owners. UC-008 weekly calls and UC-009 personal recovery are accepted. App revision16 and receiver revision7 are deployed; the receiver uses its existing single-caller diff --git a/tests/test_content.py b/tests/test_content.py index 97daf93..391ded9 100644 --- a/tests/test_content.py +++ b/tests/test_content.py @@ -16,6 +16,21 @@ def test_render_markdown_strips_script(): assert "Hi" in html +def test_supported_markdown_keeps_local_assets_and_drops_raw_html(): + html = render_markdown( + "```\nkept\n```\n\n" + "\n\n" + "\n" + "