feat: declare governed external provider state
This commit is contained in:
@@ -0,0 +1,342 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import case, func, or_, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.core.provider_governance import (
|
||||
ExternalProviderRuntimeState,
|
||||
ExternalProviderStateContext,
|
||||
)
|
||||
from govoplan_mail.backend.db.models import (
|
||||
MailBounceSource,
|
||||
MailDeliveryCommand,
|
||||
MailMailboxFolderIndex,
|
||||
MailMailboxMessageIndex,
|
||||
MailServerEndpoint,
|
||||
MailServerProfile,
|
||||
)
|
||||
|
||||
|
||||
SMTP_PROVIDER_ID = "mail.smtp_delivery"
|
||||
IMAP_PROVIDER_ID = "mail.imap_mailbox"
|
||||
_CURRENT_INDEX_WINDOW = timedelta(minutes=30)
|
||||
|
||||
|
||||
def smtp_provider_states(
|
||||
context: ExternalProviderStateContext,
|
||||
) -> tuple[ExternalProviderRuntimeState, ...]:
|
||||
return _mail_provider_states(context, protocol="smtp")
|
||||
|
||||
|
||||
def imap_provider_states(
|
||||
context: ExternalProviderStateContext,
|
||||
) -> tuple[ExternalProviderRuntimeState, ...]:
|
||||
return _mail_provider_states(context, protocol="imap")
|
||||
|
||||
|
||||
def _mail_provider_states(
|
||||
context: ExternalProviderStateContext,
|
||||
*,
|
||||
protocol: str,
|
||||
) -> tuple[ExternalProviderRuntimeState, ...]:
|
||||
if not isinstance(context.session, Session):
|
||||
raise RuntimeError("Mail provider state requires a database session.")
|
||||
profiles = _profiles(context)
|
||||
if not profiles:
|
||||
return ()
|
||||
profile_ids = tuple(item.id for item in profiles)
|
||||
endpoints = _endpoints(context.session, profile_ids=profile_ids, protocol=protocol)
|
||||
endpoints_by_profile: dict[str, list[MailServerEndpoint]] = defaultdict(list)
|
||||
for endpoint in endpoints:
|
||||
endpoints_by_profile[endpoint.profile_id].append(endpoint)
|
||||
|
||||
observed_at = datetime.now(UTC)
|
||||
if protocol == "smtp":
|
||||
metrics = _smtp_metrics(context.session, profile_ids=profile_ids)
|
||||
return tuple(
|
||||
_smtp_state(
|
||||
profile,
|
||||
endpoints=endpoints_by_profile.get(profile.id, []),
|
||||
metrics=metrics.get(profile.id, {}),
|
||||
observed_at=observed_at,
|
||||
)
|
||||
for profile in profiles
|
||||
if endpoints_by_profile.get(profile.id) or _legacy_configured(profile, "smtp")
|
||||
)
|
||||
|
||||
metrics = _imap_metrics(context.session, profile_ids=profile_ids)
|
||||
return tuple(
|
||||
_imap_state(
|
||||
profile,
|
||||
endpoints=endpoints_by_profile.get(profile.id, []),
|
||||
metrics=metrics.get(profile.id, {}),
|
||||
observed_at=observed_at,
|
||||
)
|
||||
for profile in profiles
|
||||
if endpoints_by_profile.get(profile.id) or _legacy_configured(profile, "imap")
|
||||
)
|
||||
|
||||
|
||||
def _profiles(context: ExternalProviderStateContext) -> tuple[MailServerProfile, ...]:
|
||||
statement = select(MailServerProfile)
|
||||
if context.tenant_id is not None:
|
||||
statement = statement.where(
|
||||
or_(
|
||||
MailServerProfile.tenant_id.is_(None),
|
||||
MailServerProfile.tenant_id == context.tenant_id,
|
||||
)
|
||||
)
|
||||
return tuple(
|
||||
context.session.scalars(
|
||||
statement.order_by(
|
||||
MailServerProfile.tenant_id,
|
||||
MailServerProfile.id,
|
||||
).limit(context.max_items + 1)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _endpoints(
|
||||
session: Session,
|
||||
*,
|
||||
profile_ids: tuple[str, ...],
|
||||
protocol: str,
|
||||
) -> tuple[MailServerEndpoint, ...]:
|
||||
return tuple(
|
||||
session.scalars(
|
||||
select(MailServerEndpoint).where(
|
||||
MailServerEndpoint.profile_id.in_(profile_ids),
|
||||
MailServerEndpoint.protocol == protocol,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _smtp_metrics(
|
||||
session: Session,
|
||||
*,
|
||||
profile_ids: tuple[str, ...],
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
result: dict[str, dict[str, Any]] = defaultdict(dict)
|
||||
rows = session.execute(
|
||||
select(
|
||||
MailDeliveryCommand.profile_id,
|
||||
MailDeliveryCommand.status,
|
||||
func.count(MailDeliveryCommand.id),
|
||||
func.max(MailDeliveryCommand.completed_at),
|
||||
)
|
||||
.where(MailDeliveryCommand.profile_id.in_(profile_ids))
|
||||
.group_by(MailDeliveryCommand.profile_id, MailDeliveryCommand.status)
|
||||
)
|
||||
for profile_id, status, count, last_completed_at in rows:
|
||||
item = result[str(profile_id)]
|
||||
item[str(status)] = int(count)
|
||||
if status in {"accepted", "reconciled_accepted", "partially_refused"}:
|
||||
current = _aware(item.get("last_success_at"))
|
||||
candidate = _aware(last_completed_at)
|
||||
if candidate is not None and (current is None or candidate > current):
|
||||
item["last_success_at"] = candidate
|
||||
return result
|
||||
|
||||
|
||||
def _imap_metrics(
|
||||
session: Session,
|
||||
*,
|
||||
profile_ids: tuple[str, ...],
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
result: dict[str, dict[str, Any]] = defaultdict(dict)
|
||||
folder_rows = session.execute(
|
||||
select(
|
||||
MailMailboxFolderIndex.profile_id,
|
||||
func.count(MailMailboxFolderIndex.id),
|
||||
func.max(MailMailboxFolderIndex.indexed_at),
|
||||
)
|
||||
.where(MailMailboxFolderIndex.profile_id.in_(profile_ids))
|
||||
.group_by(MailMailboxFolderIndex.profile_id)
|
||||
)
|
||||
for profile_id, count, indexed_at in folder_rows:
|
||||
result[str(profile_id)]["indexed_folders"] = int(count)
|
||||
result[str(profile_id)]["last_indexed_at"] = _aware(indexed_at)
|
||||
message_rows = session.execute(
|
||||
select(
|
||||
MailMailboxMessageIndex.profile_id,
|
||||
func.count(MailMailboxMessageIndex.id),
|
||||
)
|
||||
.where(MailMailboxMessageIndex.profile_id.in_(profile_ids))
|
||||
.group_by(MailMailboxMessageIndex.profile_id)
|
||||
)
|
||||
for profile_id, count in message_rows:
|
||||
result[str(profile_id)]["indexed_messages"] = int(count)
|
||||
bounce_rows = session.execute(
|
||||
select(
|
||||
MailBounceSource.profile_id,
|
||||
func.count(MailBounceSource.id),
|
||||
func.sum(
|
||||
case((MailBounceSource.last_error.is_not(None), 1), else_=0)
|
||||
),
|
||||
func.max(MailBounceSource.last_success_at),
|
||||
)
|
||||
.where(
|
||||
MailBounceSource.profile_id.in_(profile_ids),
|
||||
MailBounceSource.is_active.is_(True),
|
||||
)
|
||||
.group_by(MailBounceSource.profile_id)
|
||||
)
|
||||
for profile_id, count, errors, last_success_at in bounce_rows:
|
||||
item = result[str(profile_id)]
|
||||
item["active_bounce_sources"] = int(count)
|
||||
item["bounce_source_errors"] = int(errors or 0)
|
||||
item["last_bounce_success_at"] = _aware(last_success_at)
|
||||
return result
|
||||
|
||||
|
||||
def _smtp_state(
|
||||
profile: MailServerProfile,
|
||||
*,
|
||||
endpoints: list[MailServerEndpoint],
|
||||
metrics: dict[str, Any],
|
||||
observed_at: datetime,
|
||||
) -> ExternalProviderRuntimeState:
|
||||
active = bool(profile.is_active) and (
|
||||
any(item.is_active for item in endpoints)
|
||||
or (not endpoints and _legacy_configured(profile, "smtp"))
|
||||
)
|
||||
outcome_unknown = int(metrics.get("outcome_unknown", 0))
|
||||
last_success = _aware(metrics.get("last_success_at"))
|
||||
health = (
|
||||
"inactive"
|
||||
if not active
|
||||
else "warning"
|
||||
if outcome_unknown
|
||||
else "healthy"
|
||||
if last_success is not None
|
||||
else "unknown"
|
||||
)
|
||||
return ExternalProviderRuntimeState(
|
||||
provider_id=SMTP_PROVIDER_ID,
|
||||
binding_ref=f"mail:profile:{profile.id}:smtp",
|
||||
authority_mode="governance_overlay",
|
||||
observed_at=observed_at,
|
||||
configured=True,
|
||||
active=active,
|
||||
health=health,
|
||||
freshness="not_applicable",
|
||||
conflict="pending" if outcome_unknown else "clear",
|
||||
recovery=(
|
||||
"not_applicable"
|
||||
if not active
|
||||
else "attention"
|
||||
if outcome_unknown or last_success is None
|
||||
else "ready"
|
||||
),
|
||||
last_success_at=last_success,
|
||||
detail=(
|
||||
"SMTP delivery is disabled."
|
||||
if not active
|
||||
else "SMTP outcomes require reconciliation."
|
||||
if outcome_unknown
|
||||
else "SMTP delivery has no retained successful observation yet."
|
||||
if last_success is None
|
||||
else "SMTP delivery evidence is available."
|
||||
),
|
||||
metrics={
|
||||
"pending_commands": int(metrics.get("pending", 0)),
|
||||
"temporary_failures": int(metrics.get("temporary_failure", 0)),
|
||||
"permanent_failures": int(metrics.get("permanent_failure", 0)),
|
||||
"outcome_unknown_commands": outcome_unknown,
|
||||
"active_endpoints": sum(1 for item in endpoints if item.is_active),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _imap_state(
|
||||
profile: MailServerProfile,
|
||||
*,
|
||||
endpoints: list[MailServerEndpoint],
|
||||
metrics: dict[str, Any],
|
||||
observed_at: datetime,
|
||||
) -> ExternalProviderRuntimeState:
|
||||
active = bool(profile.is_active) and (
|
||||
any(item.is_active for item in endpoints)
|
||||
or (not endpoints and _legacy_configured(profile, "imap"))
|
||||
)
|
||||
indexed_at = _aware(metrics.get("last_indexed_at"))
|
||||
errors = int(metrics.get("bounce_source_errors", 0))
|
||||
freshness = (
|
||||
"not_applicable"
|
||||
if not active
|
||||
else "unknown"
|
||||
if indexed_at is None
|
||||
else "current"
|
||||
if observed_at - indexed_at <= _CURRENT_INDEX_WINDOW
|
||||
else "stale"
|
||||
)
|
||||
health = (
|
||||
"inactive"
|
||||
if not active
|
||||
else "error"
|
||||
if errors
|
||||
else "healthy"
|
||||
if indexed_at is not None
|
||||
else "unknown"
|
||||
)
|
||||
return ExternalProviderRuntimeState(
|
||||
provider_id=IMAP_PROVIDER_ID,
|
||||
binding_ref=f"mail:profile:{profile.id}:imap",
|
||||
authority_mode="external_mirror",
|
||||
observed_at=observed_at,
|
||||
configured=True,
|
||||
active=active,
|
||||
health=health,
|
||||
freshness=freshness,
|
||||
conflict="not_applicable",
|
||||
recovery=(
|
||||
"not_applicable"
|
||||
if not active
|
||||
else "ready"
|
||||
if health == "healthy" and freshness == "current"
|
||||
else "attention"
|
||||
),
|
||||
last_success_at=indexed_at or _aware(metrics.get("last_bounce_success_at")),
|
||||
detail=(
|
||||
"IMAP mailbox access is disabled."
|
||||
if not active
|
||||
else "IMAP mailbox or bounce-source errors require attention."
|
||||
if errors
|
||||
else "IMAP mailbox state has not been indexed yet."
|
||||
if indexed_at is None
|
||||
else "IMAP mailbox index state is available."
|
||||
),
|
||||
metrics={
|
||||
"indexed_folders": int(metrics.get("indexed_folders", 0)),
|
||||
"indexed_messages": int(metrics.get("indexed_messages", 0)),
|
||||
"active_bounce_sources": int(metrics.get("active_bounce_sources", 0)),
|
||||
"bounce_source_errors": errors,
|
||||
"active_endpoints": sum(1 for item in endpoints if item.is_active),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _legacy_configured(profile: MailServerProfile, protocol: str) -> bool:
|
||||
value = profile.smtp_config if protocol == "smtp" else profile.imap_config
|
||||
return isinstance(value, dict) and bool(value)
|
||||
|
||||
|
||||
def _aware(value: object | None) -> datetime | None:
|
||||
if not isinstance(value, datetime):
|
||||
return None
|
||||
if value.tzinfo is None:
|
||||
return value.replace(tzinfo=UTC)
|
||||
return value.astimezone(UTC)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"IMAP_PROVIDER_ID",
|
||||
"SMTP_PROVIDER_ID",
|
||||
"imap_provider_states",
|
||||
"smtp_provider_states",
|
||||
]
|
||||
Reference in New Issue
Block a user