Add governed contact channel facts

src/govoplan_addresses/backend/service.py

@@ -43,6 +43,7 @@ from govoplan_addresses.backend.db.models import (
     AddressSyncSource,
     AddressSyncTombstone,
     Contact,
+    ContactChannelRule,
     ContactEmail,
     ContactPhone,
     ContactPostalAddress,
@@ -68,6 +69,7 @@ from govoplan_addresses.backend.schemas import (
     ContactPhonePayload,
     ContactPostalAddressPayload,
     ContactUpdateRequest,
+    ContactChannelRuleCreateRequest,
 )
 from govoplan_addresses.backend.vcard import ParsedVCard, ParsedVCardIssue, contacts_to_vcard, parse_vcards_with_issues
 
@@ -2514,6 +2516,80 @@ def get_visible_contact(session: Session, principal: ApiPrincipal, contact_id: s
     return contact
 
 
+def list_contact_channel_rules(
+    session: Session,
+    principal: ApiPrincipal,
+    contact_id: str,
+) -> list[ContactChannelRule]:
+    contact = get_visible_contact(session, principal, contact_id)
+    return (
+        session.query(ContactChannelRule)
+        .filter(ContactChannelRule.contact_id == contact.id)
+        .order_by(ContactChannelRule.created_at.desc(), ContactChannelRule.id.asc())
+        .all()
+    )
+
+
+def create_contact_channel_rule(
+    session: Session,
+    principal: ApiPrincipal,
+    contact_id: str,
+    payload: ContactChannelRuleCreateRequest,
+) -> ContactChannelRule:
+    contact = get_visible_contact(session, principal, contact_id)
+    if (
+        payload.effective_from is not None
+        and payload.effective_until is not None
+        and payload.effective_until <= payload.effective_from
+    ):
+        raise AddressBookError("Channel-rule end must be after its start.")
+    point_ids: set[str] = set()
+    if payload.channel == "email":
+        point_ids = {item.id for item in contact.emails}
+    elif payload.channel == "postal":
+        point_ids = {item.id for item in contact.postal_addresses}
+    if payload.contact_point_id and payload.contact_point_id not in point_ids:
+        raise AddressBookError(
+            "The selected contact point does not belong to this contact and channel."
+        )
+    rule = ContactChannelRule(
+        tenant_id=contact.tenant_id,
+        contact_id=contact.id,
+        channel=payload.channel,
+        purpose=_trim(payload.purpose),
+        contact_point_id=payload.contact_point_id,
+        decision=payload.decision,
+        legal_basis=_trim(payload.legal_basis),
+        evidence_ref=_trim(payload.evidence_ref),
+        reason=_trim(payload.reason),
+        preference_rank=payload.preference_rank,
+        locale=_trim(payload.locale),
+        effective_from=payload.effective_from,
+        effective_until=payload.effective_until,
+        created_by_account_id=_account_id(principal),
+        metadata_=dict(payload.metadata),
+    )
+    session.add(rule)
+    session.flush()
+    return rule
+
+
+def end_contact_channel_rule(
+    session: Session,
+    principal: ApiPrincipal,
+    rule_id: str,
+) -> ContactChannelRule:
+    rule = session.get(ContactChannelRule, rule_id)
+    if rule is None:
+        raise AddressBookError("Contact channel rule not found.")
+    get_visible_contact(session, principal, rule.contact_id)
+    now = utcnow()
+    if rule.effective_until is None or rule.effective_until > now:
+        rule.effective_until = now
+        session.add(rule)
+    return rule
+
+
 def update_contact(session: Session, principal: ApiPrincipal, contact_id: str, payload: ContactUpdateRequest) -> Contact:
     contact = get_visible_contact(session, principal, contact_id)
     _require_mutable_book(contact.address_book)