379 lines
18 KiB
Python
379 lines
18 KiB
Python
|
"""Permanent append-only vTiger archive used by Project CT."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
import asyncio
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import re
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from typing import Any, Iterable
|
||
|
|
|
||
|
|
from app.core.database import execute_insert, execute_query, execute_query_single
|
||
|
|
from app.services.vtiger_service import VTigerService, get_vtiger_service
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
# Document metadata and original file bytes are both archived before vTiger is retired.
|
||
|
|
ARCHIVE_MODULES = (
|
||
|
|
"Users", "Groups", "Roles", "Currency", "Accounts", "Contacts", "Leads", "Potentials", "Cases", "Project",
|
||
|
|
"ProjectMilestone", "ModComments", "Timelog", "Calendar", "Events", "Emails",
|
||
|
|
"SalesOrder", "Quotes", "Invoice", "PurchaseOrder", "Vendors", "ServiceContracts",
|
||
|
|
"Subscription", "Products", "Services", "Assets", "Documents",
|
||
|
|
)
|
||
|
|
RELATION_FIELD_RE = re.compile(r"(?:^|_)(?:id|ids|account|contact|parent|related|assigned_user)$", re.I)
|
||
|
|
MODULE_BATCH_SIZES = {"Emails": 20, "Invoice": 50, "SalesOrder": 50, "Subscription": 50}
|
||
|
|
|
||
|
|
|
||
|
|
def _transient_vtiger_failure(service) -> bool:
|
||
|
|
status = service.last_query_status
|
||
|
|
error_text = json.dumps(service.last_query_error or {}).upper()
|
||
|
|
return status is None or status == 429 or (isinstance(status, int) and status >= 500) or any(
|
||
|
|
token in error_text for token in ("TOO_MANY_REQUESTS", "TIMEOUT", "TEMPORAR")
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _canonical_payload(record: dict[str, Any]) -> tuple[str, str]:
|
||
|
|
raw = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)
|
||
|
|
return raw, hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def _parse_source_time(value: Any):
|
||
|
|
text = str(value or "").strip()
|
||
|
|
if not text:
|
||
|
|
return None
|
||
|
|
try:
|
||
|
|
parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
|
||
|
|
return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|
||
|
|
except ValueError:
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def _relation_values(payload: dict[str, Any]) -> Iterable[tuple[str, str]]:
|
||
|
|
for field, value in payload.items():
|
||
|
|
if not value or not RELATION_FIELD_RE.search(str(field)):
|
||
|
|
continue
|
||
|
|
values = value if isinstance(value, list) else [value]
|
||
|
|
for candidate in values:
|
||
|
|
candidate = str(candidate or "").strip()
|
||
|
|
if re.fullmatch(r"\d+x\d+", candidate):
|
||
|
|
yield str(field), candidate
|
||
|
|
|
||
|
|
|
||
|
|
def _target_module(vtiger_id: str) -> str | None:
|
||
|
|
prefix = str(vtiger_id).split("x", 1)[0]
|
||
|
|
row = execute_query_single(
|
||
|
|
"""SELECT module FROM vtiger_archive_records
|
||
|
|
WHERE split_part(vtiger_id,'x',1)=%s ORDER BY id DESC LIMIT 1""", (prefix,),
|
||
|
|
)
|
||
|
|
return str(row["module"]) if row else None
|
||
|
|
|
||
|
|
|
||
|
|
def archive_record(version_id: int, module: str, record: dict[str, Any]) -> bool:
|
||
|
|
vtiger_id = str(record.get("id") or "").strip()
|
||
|
|
if not vtiger_id:
|
||
|
|
raise ValueError(f"{module}-post mangler id")
|
||
|
|
raw, digest = _canonical_payload(record)
|
||
|
|
previous = execute_query_single(
|
||
|
|
"""SELECT revision_no, payload_sha256 FROM vtiger_archive_records
|
||
|
|
WHERE module=%s AND vtiger_id=%s ORDER BY revision_no DESC LIMIT 1""",
|
||
|
|
(module, vtiger_id),
|
||
|
|
)
|
||
|
|
if previous and previous.get("payload_sha256") == digest:
|
||
|
|
return False
|
||
|
|
revision = int((previous or {}).get("revision_no") or 0) + 1
|
||
|
|
execute_query(
|
||
|
|
"""INSERT INTO vtiger_archive_records
|
||
|
|
(version_id,module,vtiger_id,revision_no,source_created_at,source_modified_at,is_deleted,payload,payload_sha256)
|
||
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s::jsonb,%s)""",
|
||
|
|
(version_id, module, vtiger_id, revision, _parse_source_time(record.get("createdtime")),
|
||
|
|
_parse_source_time(record.get("modifiedtime")), bool(record.get("deleted")), raw, digest),
|
||
|
|
fetch=False,
|
||
|
|
)
|
||
|
|
for field, target in _relation_values(record):
|
||
|
|
execute_query(
|
||
|
|
"""INSERT INTO vtiger_archive_relations
|
||
|
|
(version_id,source_module,source_vtiger_id,field_name,target_vtiger_id,target_module)
|
||
|
|
VALUES (%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING""",
|
||
|
|
(version_id, module, vtiger_id, field, target, _target_module(target)), fetch=False,
|
||
|
|
)
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
async def archive_document_files(version_id: int) -> dict[str, int]:
|
||
|
|
"""Archive immutable Document and Email bytes via vTiger's files_retrieve API."""
|
||
|
|
documents = execute_query(
|
||
|
|
"""SELECT DISTINCT ON (module,vtiger_id) module,vtiger_id,payload
|
||
|
|
FROM vtiger_archive_records
|
||
|
|
WHERE module IN ('Documents','Emails') AND version_id<=%s AND is_deleted=false
|
||
|
|
ORDER BY module,vtiger_id,revision_no DESC""", (version_id,),
|
||
|
|
) or []
|
||
|
|
stats = {"entities": len(documents), "resources": 0, "archived": 0, "unchanged": 0, "errors": 0}
|
||
|
|
# A small hard ceiling keeps this one-time preservation fast without flooding vTiger.
|
||
|
|
semaphore = asyncio.Semaphore(6)
|
||
|
|
jobs = []
|
||
|
|
|
||
|
|
async def archive_resource(document: dict, payload: dict, resource_id: str) -> None:
|
||
|
|
async with semaphore:
|
||
|
|
previous_resource = execute_query_single(
|
||
|
|
"""SELECT id FROM vtiger_archive_files
|
||
|
|
WHERE source_module=%s AND document_vtiger_id=%s AND resource_vtiger_id=%s
|
||
|
|
ORDER BY id DESC LIMIT 1""",
|
||
|
|
(document["module"], document["vtiger_id"], resource_id),
|
||
|
|
)
|
||
|
|
if previous_resource:
|
||
|
|
stats["unchanged"] += 1
|
||
|
|
return
|
||
|
|
service = VTigerService()
|
||
|
|
result = None
|
||
|
|
for attempt in range(8):
|
||
|
|
result = await service.retrieve_file(resource_id)
|
||
|
|
if result or not _transient_vtiger_failure(service):
|
||
|
|
break
|
||
|
|
await asyncio.sleep(min(1.5 * (2 ** attempt), 30))
|
||
|
|
if not result:
|
||
|
|
stats["errors"] += 1
|
||
|
|
return
|
||
|
|
content = result.get("content") or b""
|
||
|
|
digest = hashlib.sha256(content).hexdigest()
|
||
|
|
filename = str(result.get("filename") or payload.get("filename") or resource_id)
|
||
|
|
execute_query(
|
||
|
|
"""INSERT INTO vtiger_archive_files
|
||
|
|
(version_id,source_module,document_vtiger_id,resource_vtiger_id,filename,content_type,size_bytes,content_sha256,content)
|
||
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING""",
|
||
|
|
(version_id, document["module"], document["vtiger_id"], resource_id, filename,
|
||
|
|
result.get("filetype") or payload.get("filetype"), len(content), digest, content),
|
||
|
|
fetch=False,
|
||
|
|
)
|
||
|
|
stats["archived"] += 1
|
||
|
|
await asyncio.sleep(0.35)
|
||
|
|
|
||
|
|
for document in documents:
|
||
|
|
payload = document.get("payload") or {}
|
||
|
|
resource_ids = re.findall(r"\d+x\d+", str(payload.get("imageattachmentids") or ""))
|
||
|
|
for resource_id in resource_ids:
|
||
|
|
stats["resources"] += 1
|
||
|
|
jobs.append(archive_resource(document, payload, resource_id))
|
||
|
|
if jobs:
|
||
|
|
await asyncio.gather(*jobs)
|
||
|
|
return stats
|
||
|
|
|
||
|
|
|
||
|
|
async def _fetch_module(module: str, modified_after=None) -> list[dict[str, Any]]:
|
||
|
|
service = get_vtiger_service()
|
||
|
|
records: list[dict[str, Any]] = []
|
||
|
|
seen_ids: set[str] = set()
|
||
|
|
offset = 0
|
||
|
|
batch_size = MODULE_BATCH_SIZES.get(module, 100)
|
||
|
|
while True:
|
||
|
|
clauses = []
|
||
|
|
if modified_after:
|
||
|
|
clauses.append(f"modifiedtime >= '{modified_after.isoformat()}'")
|
||
|
|
where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
|
||
|
|
# This vTiger deployment paginates reliably with LIMIT offset,count.
|
||
|
|
# ID-watermarks skip records in modules whose entity IDs are compared as text.
|
||
|
|
query = f"SELECT * FROM {module}{where} ORDER BY id ASC LIMIT {offset}, {batch_size};"
|
||
|
|
batch = []
|
||
|
|
for attempt in range(8):
|
||
|
|
batch = await service.query(query)
|
||
|
|
if not _transient_vtiger_failure(service):
|
||
|
|
break
|
||
|
|
await asyncio.sleep(min(1.5 * (2 ** attempt), 30))
|
||
|
|
if not batch:
|
||
|
|
if service.last_query_error or (service.last_query_status not in (None, 200)):
|
||
|
|
raise RuntimeError(f"vTiger kunne ikke hente {module}: {service.last_query_error or service.last_query_status}")
|
||
|
|
break
|
||
|
|
fresh = [row for row in batch if str(row.get("id") or "") not in seen_ids]
|
||
|
|
if not fresh:
|
||
|
|
break
|
||
|
|
records.extend(fresh)
|
||
|
|
seen_ids.update(str(row.get("id") or "") for row in fresh)
|
||
|
|
# Payload-heavy modules may return fewer than the requested 100 records
|
||
|
|
# due to a response-size cap even though later offsets still exist.
|
||
|
|
offset += len(batch)
|
||
|
|
await asyncio.sleep(0.35)
|
||
|
|
return records
|
||
|
|
|
||
|
|
|
||
|
|
async def _source_module_count(module: str) -> int:
|
||
|
|
service = get_vtiger_service()
|
||
|
|
rows = []
|
||
|
|
for attempt in range(8):
|
||
|
|
rows = await service.query(f"SELECT count(*) FROM {module};")
|
||
|
|
if not _transient_vtiger_failure(service):
|
||
|
|
break
|
||
|
|
await asyncio.sleep(min(1.5 * (2 ** attempt), 30))
|
||
|
|
if not rows or rows[0].get("count") is None:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"vTiger kunne ikke kontrollere antal poster i {module}: "
|
||
|
|
f"{service.last_query_error or service.last_query_status}"
|
||
|
|
)
|
||
|
|
return int(rows[0]["count"])
|
||
|
|
|
||
|
|
|
||
|
|
async def _close_relation_targets(version_id: int) -> dict[str, int]:
|
||
|
|
"""Fetch referenced records omitted from normal module lists (notably inactive users)."""
|
||
|
|
unresolved = execute_query(
|
||
|
|
"""SELECT DISTINCT r.target_vtiger_id
|
||
|
|
FROM vtiger_archive_relations r
|
||
|
|
WHERE r.version_id <= %s AND NOT EXISTS (
|
||
|
|
SELECT 1 FROM vtiger_archive_records t
|
||
|
|
WHERE t.vtiger_id=r.target_vtiger_id AND t.version_id <= %s
|
||
|
|
) ORDER BY r.target_vtiger_id""", (version_id, version_id),
|
||
|
|
) or []
|
||
|
|
prefix_rows = execute_query(
|
||
|
|
"""SELECT split_part(vtiger_id,'x',1) AS prefix,MIN(module) AS module
|
||
|
|
FROM vtiger_archive_records GROUP BY 1"""
|
||
|
|
) or []
|
||
|
|
module_by_prefix = {str(row["prefix"]): str(row["module"]) for row in prefix_rows}
|
||
|
|
module_by_prefix.update({"19": "Users", "21": "Currency", "53": "Roles"})
|
||
|
|
service = get_vtiger_service()
|
||
|
|
stats = {"requested": 0, "archived": 0, "not_found": 0, "errors": 0}
|
||
|
|
for item in unresolved:
|
||
|
|
target_id = str(item.get("target_vtiger_id") or "")
|
||
|
|
if not re.fullmatch(r"\d+x\d+", target_id):
|
||
|
|
continue
|
||
|
|
module = module_by_prefix.get(target_id.split("x", 1)[0])
|
||
|
|
if not module:
|
||
|
|
continue
|
||
|
|
stats["requested"] += 1
|
||
|
|
rows = []
|
||
|
|
for attempt in range(8):
|
||
|
|
rows = await service.query(f"SELECT * FROM {module} WHERE id='{target_id}' LIMIT 1;")
|
||
|
|
if not _transient_vtiger_failure(service):
|
||
|
|
break
|
||
|
|
await asyncio.sleep(min(1.5 * (2 ** attempt), 30))
|
||
|
|
if rows:
|
||
|
|
stats["archived"] += int(archive_record(version_id, module, rows[0]))
|
||
|
|
elif _transient_vtiger_failure(service):
|
||
|
|
stats["errors"] += 1
|
||
|
|
else:
|
||
|
|
stats["not_found"] += 1
|
||
|
|
await asyncio.sleep(0.35)
|
||
|
|
return stats
|
||
|
|
|
||
|
|
|
||
|
|
async def run_archive_sync(sync_kind: str, initiated_by: int | None = None) -> dict[str, Any]:
|
||
|
|
if sync_kind not in {"full", "incremental", "final"}:
|
||
|
|
raise ValueError("sync_kind skal være full, incremental eller final")
|
||
|
|
running = execute_query_single(
|
||
|
|
"""SELECT id,started_at FROM vtiger_archive_versions
|
||
|
|
WHERE status='running' ORDER BY id DESC LIMIT 1"""
|
||
|
|
)
|
||
|
|
if running:
|
||
|
|
age = datetime.now(timezone.utc) - running["started_at"]
|
||
|
|
if age.total_seconds() < 12 * 3600:
|
||
|
|
raise RuntimeError(f"Arkivsync version {running['id']} kører allerede")
|
||
|
|
execute_query(
|
||
|
|
"""UPDATE vtiger_archive_versions SET status='failed',completed_at=NOW(),
|
||
|
|
critical_errors='[{"error":"Stale sync blev lukket automatisk"}]'::jsonb WHERE id=%s""",
|
||
|
|
(running["id"],), fetch=False,
|
||
|
|
)
|
||
|
|
version_id = int(execute_insert(
|
||
|
|
"""INSERT INTO vtiger_archive_versions(sync_kind,initiated_by,source_cutoff)
|
||
|
|
VALUES (%s,%s,NOW()) RETURNING id""", (sync_kind, initiated_by),
|
||
|
|
))
|
||
|
|
counts: dict[str, Any] = {}
|
||
|
|
errors: list[dict[str, str]] = []
|
||
|
|
try:
|
||
|
|
for module in ARCHIVE_MODULES:
|
||
|
|
await asyncio.sleep(0.45)
|
||
|
|
checkpoint = execute_query_single(
|
||
|
|
"SELECT last_modified_at FROM vtiger_archive_checkpoints WHERE module=%s", (module,),
|
||
|
|
)
|
||
|
|
watermark = checkpoint.get("last_modified_at") if checkpoint and sync_kind == "incremental" else None
|
||
|
|
try:
|
||
|
|
source_total = await _source_module_count(module) if sync_kind in {"full", "final"} else None
|
||
|
|
records = await _fetch_module(module, watermark)
|
||
|
|
inserted = sum(1 for record in records if archive_record(version_id, module, record))
|
||
|
|
counts[module] = {"fetched": len(records), "new_revisions": inserted}
|
||
|
|
if source_total is not None:
|
||
|
|
counts[module]["source_total"] = source_total
|
||
|
|
if len(records) != source_total:
|
||
|
|
raise RuntimeError(f"Paritetsfejl: vTiger={source_total}, hentet={len(records)}")
|
||
|
|
source_times = [parsed for parsed in (_parse_source_time(r.get("modifiedtime")) for r in records) if parsed]
|
||
|
|
newest = max(source_times, default=None)
|
||
|
|
execute_query(
|
||
|
|
"""INSERT INTO vtiger_archive_checkpoints(module,last_modified_at,last_vtiger_id,last_successful_version_id)
|
||
|
|
VALUES (%s,%s,%s,%s) ON CONFLICT(module) DO UPDATE SET
|
||
|
|
last_modified_at=COALESCE(EXCLUDED.last_modified_at,vtiger_archive_checkpoints.last_modified_at),
|
||
|
|
last_vtiger_id=EXCLUDED.last_vtiger_id,last_successful_version_id=EXCLUDED.last_successful_version_id,
|
||
|
|
updated_at=NOW()""",
|
||
|
|
(module, newest, str(records[-1].get("id")) if records else None, version_id), fetch=False,
|
||
|
|
)
|
||
|
|
except Exception as exc:
|
||
|
|
logger.exception("Project CT sync failed for %s", module)
|
||
|
|
errors.append({"module": module, "error": str(exc)})
|
||
|
|
document_files = await archive_document_files(version_id) if sync_kind in {"full", "final"} else {
|
||
|
|
"entities": 0, "resources": 0, "archived": 0, "unchanged": 0, "errors": 0,
|
||
|
|
}
|
||
|
|
if document_files.get("errors"):
|
||
|
|
errors.append({
|
||
|
|
"module": "DocumentFiles",
|
||
|
|
"error": f"{document_files['errors']} dokumentfiler kunne ikke arkiveres",
|
||
|
|
})
|
||
|
|
relation_backfill = await _close_relation_targets(version_id)
|
||
|
|
if relation_backfill.get("errors"):
|
||
|
|
errors.append({
|
||
|
|
"module": "relations",
|
||
|
|
"error": f"{relation_backfill['errors']} relationer kunne ikke kontrolleres mod vTiger",
|
||
|
|
})
|
||
|
|
archived_counts = execute_query(
|
||
|
|
"""SELECT module,COUNT(*) AS records FROM (
|
||
|
|
SELECT DISTINCT ON (module,vtiger_id) module,vtiger_id,is_deleted
|
||
|
|
FROM vtiger_archive_records WHERE version_id <= %s
|
||
|
|
ORDER BY module,vtiger_id,revision_no DESC
|
||
|
|
) snapshot WHERE is_deleted=false GROUP BY module ORDER BY module""", (version_id,),
|
||
|
|
) or []
|
||
|
|
unresolved = execute_query_single(
|
||
|
|
"""SELECT COUNT(*)::integer AS count FROM vtiger_archive_relations r
|
||
|
|
WHERE r.version_id <= %s AND NOT EXISTS (
|
||
|
|
SELECT 1 FROM vtiger_archive_records t
|
||
|
|
WHERE t.vtiger_id=r.target_vtiger_id AND t.version_id <= %s
|
||
|
|
)""", (version_id, version_id),
|
||
|
|
) or {"count": 0}
|
||
|
|
document_quality = execute_query_single(
|
||
|
|
"""SELECT COUNT(*)::integer AS total,
|
||
|
|
COUNT(*) FILTER (WHERE COALESCE(payload->>'filename',payload->>'notes_title','')='')::integer AS missing_name
|
||
|
|
FROM (SELECT DISTINCT ON (vtiger_id) vtiger_id,payload,is_deleted
|
||
|
|
FROM vtiger_archive_records WHERE module='Documents' AND version_id <= %s
|
||
|
|
ORDER BY vtiger_id,revision_no DESC) d WHERE is_deleted=false""", (version_id,),
|
||
|
|
) or {"total": 0, "missing_name": 0}
|
||
|
|
report = {
|
||
|
|
"modules_expected": len(ARCHIVE_MODULES), "modules_completed": len(counts),
|
||
|
|
"missing_modules": [module for module in ARCHIVE_MODULES if module not in counts],
|
||
|
|
"module_sync": counts, "archived_snapshot_counts": archived_counts,
|
||
|
|
"relations": {"unresolved": int(unresolved.get("count") or 0)},
|
||
|
|
"relation_backfill": relation_backfill,
|
||
|
|
"documents": {**document_quality, "files": document_files}, "critical_errors": errors,
|
||
|
|
}
|
||
|
|
status = "failed" if errors else "completed"
|
||
|
|
execute_query(
|
||
|
|
"""UPDATE vtiger_archive_versions SET status=%s,completed_at=NOW(),module_counts=%s::jsonb,
|
||
|
|
critical_errors=%s::jsonb,control_report=%s::jsonb WHERE id=%s""",
|
||
|
|
(status, json.dumps(counts), json.dumps(errors), json.dumps(report), version_id), fetch=False,
|
||
|
|
)
|
||
|
|
return {"version_id": version_id, "status": status, "counts": counts, "critical_errors": errors}
|
||
|
|
except Exception as exc:
|
||
|
|
execute_query(
|
||
|
|
"UPDATE vtiger_archive_versions SET status='failed',completed_at=NOW(),critical_errors=%s::jsonb WHERE id=%s",
|
||
|
|
(json.dumps([{"error": str(exc)}]), version_id), fetch=False,
|
||
|
|
)
|
||
|
|
raise
|
||
|
|
|
||
|
|
|
||
|
|
def termination_readiness() -> dict[str, Any]:
|
||
|
|
final = execute_query_single(
|
||
|
|
"""SELECT id,status,completed_at,critical_errors,control_approved_at
|
||
|
|
FROM vtiger_archive_versions WHERE sync_kind='final' ORDER BY id DESC LIMIT 1"""
|
||
|
|
)
|
||
|
|
reasons = []
|
||
|
|
if not final or final.get("status") != "completed":
|
||
|
|
reasons.append("Final sync mangler eller er ikke fuldført")
|
||
|
|
if final and final.get("critical_errors"):
|
||
|
|
reasons.append("Final sync har kritiske fejl")
|
||
|
|
if not final or not final.get("control_approved_at"):
|
||
|
|
reasons.append("Kontrolrapporten er ikke godkendt")
|
||
|
|
return {"ready": not reasons, "reasons": reasons, "final_version": final}
|