Add Forgejo edit/refresh sync path for space content (T05)
Keep git as source of truth: deep-link to Forgejo editor, manual refresh to drop cache, and optional signed push webhook for automatic invalidation.
This commit is contained in:
parent
9c10037f34
commit
3f30bcd340
9 changed files with 458 additions and 5 deletions
|
|
@ -1,12 +1,37 @@
|
|||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import logging
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib import messages
|
||||
from django.contrib.auth.decorators import login_required
|
||||
from django.http import HttpRequest, HttpResponse, HttpResponseForbidden, HttpResponseNotFound
|
||||
from django.shortcuts import render
|
||||
from django.http import (
|
||||
HttpRequest,
|
||||
HttpResponse,
|
||||
HttpResponseForbidden,
|
||||
HttpResponseNotFound,
|
||||
JsonResponse,
|
||||
)
|
||||
from django.shortcuts import redirect, render
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
from django.views.decorators.http import require_POST
|
||||
|
||||
from coulomb_social.apps.core.principal import build_principal
|
||||
|
||||
from .content import load_space_page
|
||||
from .content import (
|
||||
clear_content_cache_for_repo,
|
||||
clear_content_cache_for_space,
|
||||
forgejo_blob_url,
|
||||
forgejo_edit_url,
|
||||
forgejo_repo_url,
|
||||
load_space_page,
|
||||
)
|
||||
from .models import Space
|
||||
from .services import get_space_for_member, spaces_for_member
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@login_required
|
||||
def space_detail(request: HttpRequest, slug: str) -> HttpResponse:
|
||||
|
|
@ -30,9 +55,122 @@ def space_detail(request: HttpRequest, slug: str) -> HttpResponse:
|
|||
"display_name": principal["display_name"],
|
||||
"space": space,
|
||||
"page": rendered,
|
||||
"forgejo_repo_url": forgejo_repo_url(space),
|
||||
"forgejo_edit_url": forgejo_edit_url(space, page),
|
||||
"forgejo_blob_url": forgejo_blob_url(space, page),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@login_required
|
||||
@require_POST
|
||||
def space_refresh(request: HttpRequest, slug: str) -> HttpResponse:
|
||||
"""Drop cache for this space's Forgejo binding and re-show the page."""
|
||||
principal = build_principal(request, authz_resource_id="space_refresh")
|
||||
if not principal["authz_allow"]:
|
||||
return HttpResponseForbidden("Not authorized.")
|
||||
member = principal.get("member")
|
||||
space = get_space_for_member(member, slug)
|
||||
if space is None:
|
||||
return HttpResponseNotFound("Space not found.")
|
||||
n = clear_content_cache_for_space(space)
|
||||
messages.success(
|
||||
request,
|
||||
f"Content cache cleared ({n} entries). Fresh fetch on this page load.",
|
||||
)
|
||||
page = request.POST.get("page") or request.GET.get("page") or "index"
|
||||
from django.urls import reverse
|
||||
|
||||
target = reverse("spaces:detail", kwargs={"slug": slug})
|
||||
if page and page != "index":
|
||||
target = f"{target}?page={page}"
|
||||
return redirect(target)
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
@require_POST
|
||||
def forgejo_content_webhook(request: HttpRequest) -> JsonResponse:
|
||||
"""Forgejo/Gitea push webhook → invalidate content cache for matching spaces.
|
||||
|
||||
Auth: header ``X-Coulomb-Webhook-Secret`` or ``X-Gitea-Signature`` (HMAC-SHA256
|
||||
of body with FORGEJO_WEBHOOK_SECRET) or shared secret query (discouraged).
|
||||
Configure in Forgejo: repository → Webhooks → Gitea (JSON) → push events.
|
||||
"""
|
||||
secret = (getattr(settings, "FORGEJO_WEBHOOK_SECRET", None) or "").strip()
|
||||
if not secret:
|
||||
return JsonResponse(
|
||||
{"ok": False, "error": "webhook not configured"},
|
||||
status=503,
|
||||
)
|
||||
if not _webhook_authorized(request, secret):
|
||||
return JsonResponse({"ok": False, "error": "unauthorized"}, status=401)
|
||||
|
||||
try:
|
||||
payload = json.loads(request.body.decode("utf-8") or "{}")
|
||||
except json.JSONDecodeError:
|
||||
return JsonResponse({"ok": False, "error": "invalid json"}, status=400)
|
||||
|
||||
owner, repo = _repo_from_push_payload(payload)
|
||||
if not owner or not repo:
|
||||
return JsonResponse({"ok": False, "error": "no repository in payload"}, status=400)
|
||||
|
||||
cleared = clear_content_cache_for_repo(owner, repo)
|
||||
matched = Space.objects.filter(
|
||||
forgejo_owner__iexact=owner,
|
||||
forgejo_repo__iexact=repo,
|
||||
is_active=True,
|
||||
).count()
|
||||
logger.info(
|
||||
"Forgejo webhook cache invalidate owner=%s repo=%s cleared=%s spaces=%s",
|
||||
owner,
|
||||
repo,
|
||||
cleared,
|
||||
matched,
|
||||
)
|
||||
return JsonResponse(
|
||||
{
|
||||
"ok": True,
|
||||
"owner": owner,
|
||||
"repo": repo,
|
||||
"cache_entries_cleared": cleared,
|
||||
"spaces_matched": matched,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def list_spaces_context(member) -> dict:
|
||||
return {"spaces": list(spaces_for_member(member))}
|
||||
|
||||
|
||||
def _webhook_authorized(request: HttpRequest, secret: str) -> bool:
|
||||
header_secret = request.headers.get("X-Coulomb-Webhook-Secret", "")
|
||||
if header_secret and hmac.compare_digest(header_secret, secret):
|
||||
return True
|
||||
# Gitea/Forgejo HMAC-SHA256 hex of body
|
||||
sig = request.headers.get("X-Gitea-Signature") or request.headers.get(
|
||||
"X-Hub-Signature-256", ""
|
||||
)
|
||||
if sig.startswith("sha256="):
|
||||
sig = sig[len("sha256=") :]
|
||||
if sig:
|
||||
digest = hmac.new(
|
||||
secret.encode("utf-8"), request.body, hashlib.sha256
|
||||
).hexdigest()
|
||||
if hmac.compare_digest(digest, sig):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _repo_from_push_payload(payload: dict) -> tuple[str, str]:
|
||||
repo = payload.get("repository") or {}
|
||||
full = (repo.get("full_name") or "").strip()
|
||||
if "/" in full:
|
||||
owner, name = full.split("/", 1)
|
||||
return owner, name
|
||||
owner = (repo.get("owner") or {})
|
||||
if isinstance(owner, dict):
|
||||
owner_name = owner.get("login") or owner.get("username") or ""
|
||||
else:
|
||||
owner_name = str(owner)
|
||||
name = repo.get("name") or ""
|
||||
return str(owner_name), str(name)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue