wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(db): split inbound truth-chain hot lookup
Some checks failed
CD Pipeline / workflow-shape (push) Has been cancelled
Details
CD Pipeline / cancel-stale-cd (push) Has been cancelled
Details
CD Pipeline / tests (push) Has been cancelled
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Your Name
2026-07-01 08:45:40 +08:00
parent a6dc806d38
commit abc512a7b3
7 changed files with 245 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
apps/api/src/services/awooop_truth_chain_service.py
View File

@@ -34,6 +34,9 @@ from src.services.operator_summary_cache import (
logger = structlog.get_logger(__name__)
_MAX_ROWS = 100
_INBOUND_LOOKUP_BRANCH_LIMIT = int(
os.getenv("AWOOOP_TRUTH_CHAIN_INBOUND_BRANCH_LIMIT", "25")
)
_JSON_TEXT_FIELDS = {"gate_result", "source_envelope"}
_QUALITY_SUMMARY_CACHE_TTL_SECONDS = int(
os.getenv("AWOOOP_QUALITY_SUMMARY_CACHE_TTL_SECONDS", "30")
@@ -130,6 +133,108 @@ async def _fetch_one(
return _clean_row(row) if row else None
def _inbound_lookup_branch_limit(limit: int) -> int:
return max(1, min(int(limit), _INBOUND_LOOKUP_BRANCH_LIMIT))
async def _fetch_inbound_conversation_event_rows(
db: Any,
*,
project_id: str,
source_id: str,
fingerprint_needle: str,
fingerprint_value: str,
limit: int = _MAX_ROWS,
) -> list[dict[str, Any]]:
columns = """
event_id,
project_id,
channel_type,
provider_event_id,
platform_subject_id,
channel_user_id,
channel_chat_id,
run_id,
content_type,
content_hash,
content_preview,
content_redacted,
redaction_version,
source_envelope,
attachment_sha256,
is_duplicate,
provider_ts,
received_at
"""
branch_limit = _inbound_lookup_branch_limit(limit)
rows_by_id: dict[str, dict[str, Any]] = {}
source_ref_paths = (
"{source_refs,event_ids}",
"{source_refs,incident_ids}",
"{source_refs,approval_ids}",
"{source_refs,alert_ids}",
"{source_refs,sentry_issue_ids}",
"{source_refs,signoz_alerts}",
)
async def fetch_branch(where_sql: str, params: dict[str, Any]) -> None:
if len(rows_by_id) >= limit:
return
rows = await _fetch_all(
db,
f"""
SELECT
{columns}
FROM awooop_conversation_event
WHERE project_id = :project_id
AND ({where_sql})
ORDER BY received_at DESC
LIMIT :limit
""",
{
"project_id": project_id,
"source_id": source_id,
"limit": min(branch_limit, limit - len(rows_by_id)),
**params,
},
)
for row in rows:
event_id = str(row.get("event_id") or "")
if event_id:
rows_by_id.setdefault(event_id, row)
await fetch_branch("run_id::text = :source_id", {})
await fetch_branch("provider_event_id = :source_id", {})
for source_ref_path in source_ref_paths:
await fetch_branch(
f"coalesce(source_envelope #> '{source_ref_path}', '[]'::jsonb) ? :source_id",
{},
)
if fingerprint_value:
await fetch_branch(
"coalesce(source_envelope #> '{source_refs,fingerprints}', '[]'::jsonb) ? :fingerprint_value",
{"fingerprint_value": fingerprint_value},
)
if len(rows_by_id) < branch_limit:
await fetch_branch(
"content_preview ILIKE :source_needle",
{"source_needle": f"%{source_id}%"},
)
if fingerprint_needle and len(rows_by_id) < branch_limit:
await fetch_branch(
"provider_event_id ILIKE :fingerprint_needle OR content_preview ILIKE :fingerprint_needle",
{"fingerprint_needle": fingerprint_needle},
)
return sorted(
rows_by_id.values(),
key=lambda row: str(row.get("received_at") or ""),
reverse=True,
)[:limit]
def _source_type(source_id: str, incident: dict[str, Any] | None, drift: dict[str, Any] | None) -> str:
if incident is not None:
return "incident"
@@ -1745,60 +1850,13 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
else ""
)
fingerprint_value = incident_fingerprints[0] if incident_fingerprints else ""
inbound_rows = await _fetch_all(
inbound_rows = await _fetch_inbound_conversation_event_rows(
db,
"""
SELECT
event_id,
project_id,
channel_type,
provider_event_id,
platform_subject_id,
channel_user_id,
channel_chat_id,
run_id,
content_type,
content_hash,
content_preview,
content_redacted,
redaction_version,
source_envelope,
attachment_sha256,
is_duplicate,
provider_ts,
received_at
FROM awooop_conversation_event
WHERE project_id = :project_id
AND (
run_id::text = :source_id
OR provider_event_id = :source_id
OR content_preview ILIKE :source_needle
OR coalesce(source_envelope #> '{source_refs,event_ids}', '[]'::jsonb) ? :source_id
OR coalesce(source_envelope #> '{source_refs,incident_ids}', '[]'::jsonb) ? :source_id
OR coalesce(source_envelope #> '{source_refs,approval_ids}', '[]'::jsonb) ? :source_id
OR coalesce(source_envelope #> '{source_refs,alert_ids}', '[]'::jsonb) ? :source_id
OR coalesce(source_envelope #> '{source_refs,sentry_issue_ids}', '[]'::jsonb) ? :source_id
OR coalesce(source_envelope #> '{source_refs,signoz_alerts}', '[]'::jsonb) ? :source_id
OR (
:fingerprint_needle != ''
AND (
provider_event_id ILIKE :fingerprint_needle
OR content_preview ILIKE :fingerprint_needle
OR coalesce(source_envelope #> '{source_refs,fingerprints}', '[]'::jsonb) ? :fingerprint_value
)
)
)
ORDER BY received_at DESC
LIMIT :limit
""",
{
"source_id": source_id,
"project_id": project_id,
"source_needle": f"%{source_id}%",
"fingerprint_needle": fingerprint_needle,
"fingerprint_value": fingerprint_value,
"limit": _MAX_ROWS,
},
project_id=project_id,
source_id=source_id,
fingerprint_needle=fingerprint_needle,
fingerprint_value=fingerprint_value,
limit=_MAX_ROWS,
)
outbound_rows = await _fetch_all(
db,