"""Form Version History — session-authenticated routes for the edit_site UI.
These routes use login_required (session auth) rather than API key auth,
since they serve the web UI on /sites/{id}/edit (Versions tab).
"""
from functools import wraps
from flask import Blueprint, flash, g, jsonify, redirect, request, session, url_for
from app.auth_utils import login_required
from app.models import (
get_db,
get_site_version,
get_site_versions,
get_user_tier,
get_version_limit,
parse_site_fields,
rollback_site_version,
save_site_version,
)
versions_bp = Blueprint("versions", __name__, url_prefix="/versions")
def require_site_access(site_id):
"""Ensure the logged-in user owns or is a team member with access to the site."""
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
conn = get_db()
try:
site = conn.execute(
"SELECT * FROM sites WHERE id = ? AND (user_id = ? OR team_id IN (SELECT team_id FROM team_members WHERE user_id = ? AND status = 'active'))",
(site_id, session["user_id"], session["user_id"]),
).fetchone()
if site is None:
flash("Site not found or access denied", "error")
return redirect(url_for("user_sites.list_sites"))
g.site = site
return f(*args, **kwargs)
finally:
conn.close()
return decorated
return decorator
def compute_changes(older_fields, newer_fields):
"""Compute field-level diff between two version snapshots."""
old_map = {f.get("key", ""): f for f in older_fields}
new_map = {f.get("key", ""): f for f in newer_fields}
all_keys = set(list(old_map.keys()) + list(new_map.keys()))
changes = {}
for key in all_keys:
in_new = key in new_map
in_old = key in old_map
if in_new and not in_old:
changes[key] = {"type": "added", "old": None, "new": new_map[key]}
elif in_old and not in_new:
changes[key] = {"type": "removed", "old": old_map[key], "new": None}
elif new_map[key] != old_map[key]:
changes[key] = {"type": "modified", "old": old_map[key], "new": new_map[key]}
return changes
@versions_bp.route("/<int:site_id>/history", methods=["GET"], endpoint="get_history")
@login_required
@require_site_access
def get_history(site_id):
"""GET /versions/{site_id}/history — list version history for a site.
Returns JSON array of versions (newest first) with metadata for the timeline UI.
Each version is enriched with: changes (diff from previous), current flag, reason alias.
"""
site = g.site
tier, _ = get_user_tier(site.get("user_id"))
limit = get_version_limit(tier)
versions = get_site_versions(site["id"], limit=limit)
# Sort by version ascending so we can compute diffs chronologically
versions_sorted = sorted(versions, key=lambda v: v["version"])
# Current site field config
current_fields = parse_site_fields(site) or []
enriched = []
for i, v in enumerate(versions_sorted):
# Compute changes vs previous version
if i == 0:
prev_fields = []
else:
prev_fields = versions_sorted[i - 1].get("field_config_parsed", []) or []
curr_fields = v.get("field_config_parsed", []) or []
changes = compute_changes(prev_fields, curr_fields)
# Check if this version matches current site state
is_current = curr_fields == current_fields
enriched.append(
{
"id": v.get("id"),
"version": v["version"],
"created_at": v.get("changed_at", ""),
"reason": v.get("change_reason", ""),
"changed_by": v.get("changed_by", ""),
"current": is_current,
"changes": changes if changes else None,
}
)
# Return newest first
enriched.reverse()
return jsonify(
{
"site_id": site["id"],
"site_name": site["name"],
"versions": enriched,
"current_fields": current_fields,
"count": len(enriched),
}
)
@versions_bp.route("/<int:site_id>/versions/<int:version_id>", methods=["GET"], endpoint="view_version")
@login_required
@require_site_access
def view_version(site_id, version_id):
"""GET /versions/{site_id}/versions/{version_id} — get a specific version.
Returns the full version record including field_config for diff display.
"""
version = get_site_version(site_id, version_id)
if not version:
return jsonify({"error": "Version not found"}), 404
return jsonify(
{
"version": version,
}
)
@versions_bp.route("/<int:site_id>/snapshot", methods=["POST"], endpoint="create_snapshot")
@login_required
@require_site_access
def create_snapshot(site_id):
"""POST /versions/{site_id}/snapshot — manually save current config as a version.
Request body: { "reason": "optional reason" }
"""
data = request.get_json(silent=True) or {}
reason = data.get("reason", "Manual snapshot")
save_site_version(
site_id=site_id,
changed_by=session.get("email", "user"),
change_reason=reason,
)
return jsonify(
{
"saved": True,
"site_id": site_id,
}
)
@versions_bp.route("/<int:site_id>/rollback/<int:version_id>", methods=["POST"], endpoint="rollback")
@login_required
@require_site_access
def rollback(site_id, version_id):
"""POST /versions/{site_id}/rollback/{version_id} — revert to a previous version.
Request body: { "reason": "optional reason" }
"""
data = request.get_json(silent=True) or {}
reason = data.get("reason", "Manual rollback")
target = get_site_version(site_id, version_id)
if not target:
return jsonify({"error": "Version not found"}), 404
new_version = rollback_site_version(
site_id=site_id,
version_id=version_id,
changed_by=session.get("email", "user"),
change_reason=reason,
)
if new_version is None:
return jsonify({"error": "Rollback failed"}), 500
return jsonify(
{
"rolled_back": True,
"new_version": new_version,
"restored_from": version_id,
}
)
@versions_bp.route("/<int:site_id>/compare/<int:newer>/<int:older>", methods=["GET"], endpoint="compare_versions")
@login_required
@require_site_access
def compare_versions(site_id, newer, older):
"""GET /versions/{site_id}/compare/{newer}/{older} — compare two versions.
Returns field-level diff between two version snapshots.
"""
v_new = get_site_version(site_id, newer)
v_old = get_site_version(site_id, older)
if not v_new:
return jsonify({"error": "Newer version not found"}), 404
if not v_old:
return jsonify({"error": "Older version not found"}), 404
new_fields = v_new.get("field_config_parsed", []) or []
old_fields = v_old.get("field_config_parsed", []) or []
changes = compute_changes(old_fields, new_fields)
return jsonify(
{
"newer": newer,
"older": older,
"changes": changes,
"total_changes": len(changes),
}
)