"""Campaigns: create, preview, dispatch, acknowledge, report. Three things in here carry the design weight. **Sending is one transaction per phase, never one across the fan-out.** The flip to ``sending`` commits before anything external happens, target rows commit before the first envelope is built, and the fan-out itself commits per recipient (``fanout.enqueue_then_drain``). A transaction held open across 500 Feishu calls is how this platform previously locked the login path behind an outbox write, and how an idle connection collects a 2013. **The flip to ``sending`` IS the lock.** It is a conditional UPDATE on ``status``; a second dispatch tick that lands on the same row updates zero rows and returns. No advisory lock, no lease column — the state machine already had the information. **Delivery truth is read from the outbox, never mirrored.** The target row records what *we* did (queued / opened / acknowledged). Whether the message actually landed belongs to ``notification_outbox``, which owns the retry ladder, the ``dead`` terminal state and the leaver gate. """ from __future__ import annotations import contextlib import json import logging import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import Select, func, select, update from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.core.person_filters import FilterError from app.domains.security_awareness.audience import ( AudienceMember, AudienceResolver, ) from app.domains.security_awareness.catalog import ( ACK_SOURCES, ACK_WEB, CAMPAIGN_CANCELLED, CAMPAIGN_CHANNELS, CAMPAIGN_CLOSED, CAMPAIGN_DRAFT, CAMPAIGN_LIVE_STATUSES, CAMPAIGN_RESOURCE_TYPE, CAMPAIGN_SCHEDULED, CAMPAIGN_SENDING, CAMPAIGN_SENT, COMPLETION_ACK, COMPLETION_IMPRESSION, COMPLETION_OPEN, MAX_AUDIENCE, STATUS_PUBLISHED, TEMPLATE_MANDATORY, TEMPLATE_NOTICE, ) from app.domains.security_awareness.models import ( SecawCampaign, SecawCampaignTarget, SecawMaterial, SecawMaterialQuestion, ) from app.domains.security_awareness.service import ( estimate_minutes, pick_block, ) from app.identity.person_models import IdentityPerson from app.notifications.models import ( STATUS_DELIVERED, STATUS_PARTIAL, NotificationOutbox, ) logger = logging.getLogger(__name__) #: How long a campaign may sit in ``sending`` before the dispatcher treats #: it as a dead run and re-arms it. #: #: ``sending`` is now a SHORT state — ``_dispatch_inner`` settles the #: status the moment the target ledger is durable, which is ~20s for a #: 400-person audience and bounded by the ``MAX_AUDIENCE`` cap — so a row #: still in it a quarter of an hour later is a process that died mid #: phase 1, not a slow one. Before this ceiling existed nothing in the #: codebase ever moved a campaign out of ``sending``: ``dispatch_due`` #: only ever selected ``scheduled``, so a wedged campaign stayed wedged #: forever, its coverage frozen at zero and ``close()`` permanently #: refusing it (2026-09-07). STALE_SENDING_AFTER: timedelta = timedelta(minutes=15) #: How close two identical campaigns' send moments must be for arming the #: second one to be refused as an accidental double send. #: #: The shape this guards: the send used to run the whole fan-out inside #: the HTTP request, so a company-wide campaign outlived nginx's 300s #: ``proxy_read_timeout`` and the drawer showed a failure while the send #: was in fact still running. The natural next move — reopen the drawer, #: press the button again — does NOT retry anything: it POSTs #: ``create`` a second time, so 400 people get the same required reading #: twice and the coverage report grows a phantom campaign. The drain is #: off-request now, but a mis-click, an impatient double submit and a #: retried automation all produce the same request, and only the server #: can refuse it. DUPLICATE_SEND_WINDOW: timedelta = timedelta(minutes=30) ADMIN_MAX_PAGE_SIZE = 100 TARGET_MAX_PAGE_SIZE = 200 #: Outbox statuses that mean the message reached the person. ``partial`` #: is included deliberately — one channel landing is reach. _DELIVERED_STATUSES: tuple[str, ...] = (STATUS_DELIVERED, STATUS_PARTIAL) #: Employee-facing route the card / mail button points at. The campaign #: does NOT get its own landing page: the thing the reader is being sent #: to IS the material, and a campaign-shaped intermediate page would be a #: click between the push and the content. The ack control renders on the #: material page itself when the reader owes an acknowledgement. MATERIAL_ROUTE = "/me/security-learning/{material_id}" class CampaignService: def __init__(self, db: AsyncSession): self.db = db self.audit = AuditService(db) # ════════════════════════════════════════════════════════════════ # Reads — admin console # ════════════════════════════════════════════════════════════════ async def list_campaigns( self, *, status: str | None = None, mandatory: bool | None = None, material_id: uuid.UUID | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: page = max(1, page) page_size = max(1, min(page_size, ADMIN_MAX_PAGE_SIZE)) stmt: Select[Any] = select(SecawCampaign) if status: stmt = stmt.where(SecawCampaign.status == status) if mandatory is not None: stmt = stmt.where(SecawCampaign.mandatory.is_(mandatory)) if material_id: stmt = stmt.where(SecawCampaign.material_id == material_id) stmt = stmt.order_by(SecawCampaign.created_at.desc()) rows = list((await self.db.execute(stmt)).scalars().all()) total = len(rows) page_rows = rows[(page - 1) * page_size:(page - 1) * page_size + page_size] funnels = await self._funnel_map([c.id for c in page_rows]) titles = await self._material_titles({c.material_id for c in page_rows}) return [ { **self._campaign_row(c), "material_title": titles.get(c.material_id), "funnel": funnels.get(c.id, self._empty_funnel()), } for c in page_rows ], total async def get_campaign(self, campaign_id: uuid.UUID) -> dict[str, Any]: campaign = await self._require(campaign_id) titles = await self._material_titles({campaign.material_id}) funnel = (await self._funnel_map([campaign.id])).get( campaign.id, self._empty_funnel(), ) return { **self._campaign_row(campaign), "material_title": titles.get(campaign.material_id), "funnel": funnel, "by_department": await self._breakdown(campaign.id, "department"), "by_country": await self._breakdown(campaign.id, "country"), } async def list_targets( self, campaign_id: uuid.UUID, *, state: str = "all", q: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[dict[str, Any]], int]: """The per-person table, with delivery status joined from the outbox rather than mirrored onto the target row.""" await self._require(campaign_id) page = max(1, page) page_size = max(1, min(page_size, TARGET_MAX_PAGE_SIZE)) stmt = select(SecawCampaignTarget).where( SecawCampaignTarget.campaign_id == campaign_id, ) if state == "pending": stmt = stmt.where(SecawCampaignTarget.acknowledged_at.is_(None)) elif state == "acknowledged": stmt = stmt.where(SecawCampaignTarget.acknowledged_at.is_not(None)) elif state == "unopened": stmt = stmt.where(SecawCampaignTarget.opened_at.is_(None)) rows = list((await self.db.execute( stmt.order_by(SecawCampaignTarget.created_at.asc()) )).scalars().all()) if q: needle = q.strip().lower() rows = [ r for r in rows if needle in str( (r.person_snapshot or {}).get("display_name") or "", ).lower() or needle in str( (r.person_snapshot or {}).get("work_email") or "", ).lower() ] total = len(rows) page_rows = rows[(page - 1) * page_size:(page - 1) * page_size + page_size] delivery = await self._delivery_map( campaign_id, [r.person_id for r in page_rows], ) return [ { "person_id": str(r.person_id), # Name and email are separate fields end to end (#3417). "name": (r.person_snapshot or {}).get("display_name"), "email": (r.person_snapshot or {}).get("work_email"), "department": (r.person_snapshot or {}).get("department"), "country": (r.person_snapshot or {}).get("country"), "lang": r.lang, "material_version": r.material_version, "queued_at": r.queued_at, "opened_at": r.opened_at, "acknowledged_at": r.acknowledged_at, "ack_source": r.ack_source, "resend_count": r.resend_count, "last_sent_at": r.last_sent_at, # From the outbox. ``None`` = no outbox row found, which # is a real state (queue failed) and not the same as # "pending". "delivery_status": (delivery.get(r.person_id) or {}).get("status"), # WHICH channel missed, and WHY. ``delivery_status`` # alone says ``partial`` and stops there — which reads # as "mostly fine" and hides the only question an admin # actually has when somebody reports not seeing the # notice. Reconstructing it meant querying the outbox by # hand; now the row carries it. "failed_channels": ( (delivery.get(r.person_id) or {}).get("failed_channels") or [] ), "delivery_error": (delivery.get(r.person_id) or {}).get("error"), } for r in page_rows ], total async def preview_audience( self, audience_config: dict[str, Any] | None, ) -> dict[str, Any]: try: preview = await AudienceResolver(self.db).preview(audience_config) except FilterError as exc: raise AppException( code="SECAW_AUDIENCE_INVALID", message=str(exc), status_code=422, ) from exc return { "total": preview.total, "sample": preview.sample, "by_department": preview.by_department, "by_country": preview.by_country, "notes": preview.notes, "cap": MAX_AUDIENCE, } # ════════════════════════════════════════════════════════════════ # Reads — employee # ════════════════════════════════════════════════════════════════ async def my_pending( self, person_id: uuid.UUID, *, include_voluntary: bool = False, ) -> list[dict[str, Any]]: """Campaigns this person still owes. Mandatory only by default: a voluntary notice is not work, and putting it in a todo list would make the two indistinguishable — which is the whole reason the strengths are separate. """ stmt = ( select(SecawCampaignTarget, SecawCampaign, SecawMaterial) .join(SecawCampaign, SecawCampaign.id == SecawCampaignTarget.campaign_id) .join(SecawMaterial, SecawMaterial.id == SecawCampaign.material_id) .where( SecawCampaignTarget.person_id == person_id, SecawCampaignTarget.acknowledged_at.is_(None), # ``sent`` only, not the whole live set: a CLOSED # campaign has stopped asking, and leaving its rows here # would leave employees holding work nobody is waiting # for any more. SecawCampaign.status == CAMPAIGN_SENT, SecawCampaign.completion_mode == COMPLETION_ACK, ) ) if not include_voluntary: stmt = stmt.where(SecawCampaign.mandatory.is_(True)) rows = (await self.db.execute( # ``nulls_last()`` renders PostgreSQL's ``NULLS LAST``, which # MySQL / TiDB reject with a 1064. MySQL sorts NULLs FIRST on # ASC, so "soonest deadline first, undated last" needs this # explicit is-null sort key. stmt.order_by( SecawCampaign.due_at.is_(None).asc(), SecawCampaign.due_at.asc(), ) )).all() out: list[dict[str, Any]] = [] # Which of these are confirmed by answering a quiz rather than by # pressing confirm. Resolved in ONE query over the whole list # rather than per row — and surfaced because it changes what the # person is being asked to do. Without it a todo list (and the # MCP tool that reads the same rows) can only find out by trying # to confirm and being refused. quiz_material_ids: set[uuid.UUID] = set() material_ids = [material.id for _t, _c, material in rows] if material_ids: quiz_material_ids = set((await self.db.execute( select(SecawMaterialQuestion.material_id) .where(SecawMaterialQuestion.material_id.in_(material_ids)) .distinct() )).scalars().all()) for target, campaign, material in rows: try: _lang, block, _fb = pick_block(material.localized, target.lang) except ValueError: # pragma: no cover - guarded on write continue out.append({ "campaign_id": str(campaign.id), "material_id": str(material.id), "title": block.get("title"), "summary": block.get("summary"), "kind": material.kind, "category": material.category, "mandatory": campaign.mandatory, "due_at": campaign.due_at, "opened_at": target.opened_at, # True = the way to complete this is the quiz, not a # confirm button. "has_quiz": material.id in quiz_material_ids, "estimated_minutes": estimate_minutes( material.kind, block, material.estimated_minutes, ), }) return out async def pending_all_for_material( self, person_id: uuid.UUID, material_id: uuid.UUID, ) -> list[dict[str, Any]]: """Every acknowledgement this reader owes on THIS material. Usually zero or one. It can be more than one — the same material genuinely goes out twice (a second wave, a yearly re-run that did not duplicate the row) — and the plural is what the quiz path needs: one passing attempt is one demonstration of understanding of one piece of content, so it settles every campaign asking for exactly that. Settling only the most urgent would leave the other outstanding with no way left to clear it, because the quiz component will not re-offer a quiz somebody has already passed. Most urgent first (mandatory, then soonest due). """ rows = (await self.db.execute( select(SecawCampaignTarget, SecawCampaign) .join(SecawCampaign, SecawCampaign.id == SecawCampaignTarget.campaign_id) .where( SecawCampaignTarget.person_id == person_id, SecawCampaignTarget.acknowledged_at.is_(None), SecawCampaign.material_id == material_id, SecawCampaign.status == CAMPAIGN_SENT, SecawCampaign.completion_mode == COMPLETION_ACK, ) .order_by( SecawCampaign.mandatory.desc(), # Same MySQL/TiDB constraint as ``my_pending`` — see the # comment there. SecawCampaign.due_at.is_(None).asc(), SecawCampaign.due_at.asc(), ) )).all() return [ { "campaign_id": str(campaign.id), "mandatory": campaign.mandatory, "due_at": campaign.due_at, "opened_at": target.opened_at, } for target, campaign in rows ] async def pending_for_material( self, person_id: uuid.UUID, material_id: uuid.UUID, ) -> dict[str, Any] | None: """The single most urgent acknowledgement owed on this material. Drives the ack banner on the material page: one button, for the thing most in need of pressing. The plural lives in :meth:`pending_all_for_material`. """ rows = await self.pending_all_for_material(person_id, material_id) return rows[0] if rows else None # ════════════════════════════════════════════════════════════════ # Employee writes # ════════════════════════════════════════════════════════════════ async def mark_opened( self, *, person_id: uuid.UUID, material_id: uuid.UUID, ) -> None: """Stamp ``opened_at`` on every live target for this material. Called from the material read route. First-open only (a conditional UPDATE on ``opened_at IS NULL``): "when did they first open it" has one answer, and re-stamping on every refresh would turn the column into "when did they last look", which is a different claim than the one the coverage report makes. """ campaign_ids = (await self.db.execute( select(SecawCampaign.id).where( SecawCampaign.material_id == material_id, SecawCampaign.status.in_(CAMPAIGN_LIVE_STATUSES), ) )).scalars().all() if not campaign_ids: return await self.db.execute( update(SecawCampaignTarget) .where( SecawCampaignTarget.person_id == person_id, SecawCampaignTarget.campaign_id.in_(campaign_ids), SecawCampaignTarget.opened_at.is_(None), ) .values(opened_at=datetime.now(UTC)) ) async def acknowledge( self, *, person_id: uuid.UUID, campaign_id: uuid.UUID, source: str = ACK_WEB, trace_id: str | uuid.UUID = "", quiz_verified: bool = False, ) -> dict[str, Any]: """Record "I have read and understood". Idempotent: acknowledging twice returns the FIRST timestamp rather than moving it. The timestamp is the evidence; a second click must not be able to rewrite when the person actually confirmed. **When the material carries a quiz, passing it is the only way in.** The web and mobile pages already hide the confirm button in that case, but a hidden button is not a rule: this endpoint is also reachable directly and through the ``acknowledge_security_ awareness`` MCP tool, and either one would let somebody assert understanding without demonstrating it — which is the exact gap the quiz exists to close, and the reason no "answer the quiz for me" MCP tool was shipped. ``quiz_verified`` is set only by the submit path, which has just graded a passing attempt. """ if source not in ACK_SOURCES: raise AppException( code="SECAW_ACK_SOURCE_INVALID", message=f"unknown ack source {source!r}", status_code=422, ) row = (await self.db.execute( select(SecawCampaignTarget).where( SecawCampaignTarget.campaign_id == campaign_id, SecawCampaignTarget.person_id == person_id, ) )).scalar_one_or_none() if row is None: # Not "forbidden" — this campaign was never sent to them, so # from their side it does not exist. raise NotFoundError("security awareness campaign", str(campaign_id)) campaign = await self._require(campaign_id) # ``sent`` only. A CLOSED campaign has had its number reported — # the funnel at close time is in the audit row — and accepting a # late acknowledgement would silently move a figure somebody has # already cited. Closing is the act that stops the asking, so it # has to stop the answering too. if campaign.status != CAMPAIGN_SENT: raise AppException( code="SECAW_CAMPAIGN_NOT_OPEN", message="this campaign is not accepting acknowledgements", status_code=409, ) if row.acknowledged_at is not None: # Before the quiz gate on purpose: attaching a quiz to a # material AFTER a campaign went out must not retroactively # un-confirm the people who already did what was asked. return { "acknowledged_at": row.acknowledged_at, "already": True, } if not quiz_verified: from app.domains.security_awareness.quiz import MaterialQuizService quiz = MaterialQuizService(self.db) if await quiz.has_quiz(campaign.material_id) and not ( await quiz.has_passed(person_id, campaign.material_id) ): raise AppException( code="SECAW_QUIZ_REQUIRED", message=( "this material is confirmed by answering its quiz, " "not by pressing confirm" ), status_code=409, ) now = datetime.now(UTC) row.acknowledged_at = now row.ack_source = source if row.opened_at is None: # Confirming without an open is possible (the card carries # the summary, and an employee may act on it from Feishu). # Recorded as an open too, because they did engage — but the # ledger keeps the ack source so the two are distinguishable. row.opened_at = now await self.db.flush() # Settle the nag account. Constraint (3) in ``cadence``'s module # docstring: without this, this person's NEXT ask of the same # kind inherits the counter and starts on the two-week rung — a # "noise reduction" that turns a fresh problem into near-silence. from app.domains.security_awareness.reminders import clear_nag_state await clear_nag_state( self.db, person_id=person_id, campaign_id=campaign_id, ) # Read the STORED value back before returning it. The column is a # MySQL/TiDB DATETIME: it drops the microseconds and the tzinfo # this in-memory value carries, so returning ``now`` would make # the first response disagree with every later one about the same # fact — and this fact is evidence somebody may cite. await self.db.refresh(row, ["acknowledged_at"]) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=person_id, action="security_awareness.campaign.acknowledge", status="success", resource_type="secaw_campaign", resource_id=str(campaign_id), detail={"source": source, "material_id": str(campaign.material_id)}, ) return {"acknowledged_at": row.acknowledged_at, "already": False} # ════════════════════════════════════════════════════════════════ # Admin writes # ════════════════════════════════════════════════════════════════ async def create( self, *, material_id: uuid.UUID, name: str, audience_config: dict[str, Any] | None, channels: list[str], mandatory: bool, completion_mode: str, due_at: datetime | None, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> SecawCampaign: material = await self.db.get(SecawMaterial, material_id) if material is None: raise NotFoundError("security awareness material", str(material_id)) if material.status != STATUS_PUBLISHED: # A draft cannot be campaigned: the recipients would follow # the card to a 404. Publishing first is one click and is the # author confirming the content is ready to be seen. raise AppException( code="SECAW_MATERIAL_NOT_PUBLISHED", message="publish the material before sending it to anyone", status_code=400, ) # Validate the audience NOW rather than at dispatch time — a # malformed filter discovered by a cron at 09:00 is a silent # non-send, and the admin who wrote it is not watching. await self.preview_audience(audience_config) now = datetime.now(UTC) row = SecawCampaign( id=uuid.uuid4(), material_id=material_id, name=name.strip(), audience_config=audience_config or {}, channels=self._clean_channels(channels), mandatory=mandatory, completion_mode=completion_mode, status=CAMPAIGN_DRAFT, due_at=due_at, created_at=now, updated_at=now, created_by_person_id=actor_id, updated_by_person_id=actor_id, ) self.db.add(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.campaign.create", status="success", resource_type="secaw_campaign", resource_id=str(row.id), detail=self._snapshot(row), ) return row async def update( self, campaign_id: uuid.UUID, patch: dict[str, Any], *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> SecawCampaign: """Edit a campaign that has not gone out yet. Once it has sent, the audience, material and strength are part of the record — editing them would silently restate what a set of people were actually asked to do. Only ``due_at`` stays editable (extending a deadline is a real, benign operational need), and the reminder ladder in PR4 reads it live. """ campaign = await self._require(campaign_id) before = self._snapshot(campaign) sent = campaign.status not in (CAMPAIGN_DRAFT, CAMPAIGN_SCHEDULED) if sent: unexpected = set(patch) - {"due_at", "name"} if unexpected: raise AppException( code="SECAW_CAMPAIGN_ALREADY_SENT", message=( "a campaign that has been sent can only have its " f"name and due date changed (got: {sorted(unexpected)})" ), status_code=409, ) if "name" in patch and patch["name"]: campaign.name = str(patch["name"]).strip() if "due_at" in patch: campaign.due_at = patch["due_at"] if "material_id" in patch and patch["material_id"]: material = await self.db.get(SecawMaterial, patch["material_id"]) if material is None: raise NotFoundError( "security awareness material", str(patch["material_id"]), ) if material.status != STATUS_PUBLISHED: raise AppException( code="SECAW_MATERIAL_NOT_PUBLISHED", message="publish the material before sending it to anyone", status_code=400, ) campaign.material_id = material.id if "audience_config" in patch: await self.preview_audience(patch["audience_config"]) campaign.audience_config = patch["audience_config"] or {} if "channels" in patch: campaign.channels = self._clean_channels(patch["channels"]) if "mandatory" in patch and patch["mandatory"] is not None: campaign.mandatory = bool(patch["mandatory"]) if "completion_mode" in patch and patch["completion_mode"]: campaign.completion_mode = str(patch["completion_mode"]) campaign.updated_by_person_id = actor_id campaign.updated_at = datetime.now(UTC) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.campaign.update", status="success", resource_type="secaw_campaign", resource_id=str(campaign.id), detail={"before": before, "after": self._snapshot(campaign)}, ) return campaign async def schedule( self, campaign_id: uuid.UUID, when: datetime | None, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, force: bool = False, ) -> SecawCampaign: """Arm the campaign. ``when=None`` means "as soon as the next dispatch tick runs" (within ten minutes). ``force`` waives the duplicate-send guard. It exists because a deliberate second wave of the same material to the same audience inside half an hour is legitimate, if rare — a guard with no way past it is a guard people route around by editing the audience by one person, which is worse than the double send. """ campaign = await self._require(campaign_id) if campaign.status not in (CAMPAIGN_DRAFT, CAMPAIGN_SCHEDULED): raise AppException( code="SECAW_CAMPAIGN_ALREADY_SENT", message=f"cannot schedule a campaign in status {campaign.status!r}", status_code=409, ) intended_at = when or datetime.now(UTC) if not force: await self._refuse_duplicate_send(campaign, intended_at) campaign.status = CAMPAIGN_SCHEDULED campaign.scheduled_at = intended_at campaign.updated_by_person_id = actor_id campaign.updated_at = datetime.now(UTC) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.campaign.schedule", status="success", resource_type="secaw_campaign", resource_id=str(campaign.id), detail={"scheduled_at": campaign.scheduled_at.isoformat()}, ) return campaign @staticmethod def _send_identity(campaign: SecawCampaign) -> tuple[Any, ...]: """The fields that make two campaigns "the same send".""" return ( str(campaign.material_id), json.dumps(campaign.audience_config or {}, sort_keys=True), bool(campaign.mandatory), campaign.completion_mode, tuple(sorted(campaign.channels or [])), ) @staticmethod def _send_moment(campaign: SecawCampaign) -> datetime | None: """When this campaign reaches (or reached) people. ``sent_at`` once it has gone out, otherwise the moment it is armed for. Stored DATETIMEs come back naive; normalised to UTC so a subtraction against an aware ``datetime.now(UTC)`` does not raise. """ moment = campaign.sent_at or campaign.scheduled_at if moment is None: return None return moment if moment.tzinfo else moment.replace(tzinfo=UTC) async def _refuse_duplicate_send( self, campaign: SecawCampaign, intended_at: datetime, ) -> None: """409 if an identical campaign reaches the same people within :data:`DUPLICATE_SEND_WINDOW` of ``intended_at``. Matched on every field the author's form actually submits EXCEPT the name — material, audience, strength, completion mode, channels — because that is precisely what a re-submitted drawer reproduces, while any deliberately different campaign differs in at least one of them (a second wave to another department, the voluntary follow-up to a required reading). The name is excluded on purpose: keying on it would reduce the guard to "did you also retype the title", which a duplicate submit passes trivially. 🔴 **The window is between the two SEND MOMENTS, not "how recently was the twin row touched".** A first cut compared the twin's ``updated_at`` against ``now``, which was wrong twice over: renaming or re-dating an old sent campaign refreshes ``updated_at`` and would have made a months-old campaign block a legitimate new one for half an hour, while arming an identical campaign for NEXT WEEK — plainly deliberate, and the reason the scheduler takes a date at all — would have been refused because the previous wave happened twenty minutes ago. Comparing the moments people actually receive things answers the question the guard is asking. It also lets ``scheduled`` twins count, which catches arming the same send twice for the same future hour — a double send the status-based version could not see at all. The identity comparison happens in Python. ``audience_config`` is a JSON column and MySQL/TiDB JSON equality is not the same relation as "the admin filled in the same form" — key order and integer/float spelling both differ from it. The candidate set is one material's campaigns, so there is nothing to gain from pushing it into SQL. """ twins = (await self.db.execute( select(SecawCampaign) .where( SecawCampaign.id != campaign.id, SecawCampaign.material_id == campaign.material_id, SecawCampaign.status.in_( (CAMPAIGN_SCHEDULED, CAMPAIGN_SENDING, CAMPAIGN_SENT), ), ) )).scalars().all() mine = self._send_identity(campaign) for twin in twins: if self._send_identity(twin) != mine: continue moment = self._send_moment(twin) if moment is None: continue if abs(intended_at - moment) >= DUPLICATE_SEND_WINDOW: continue verb = "already sent" if twin.sent_at else "is already armed to send" raise AppException( code="SECAW_CAMPAIGN_DUPLICATE_SEND", message=( f"“{twin.name}” {verb} this material to the same " f"audience at {moment:%Y-%m-%d %H:%M} UTC. Sending " "again would give every one of those people a second " "copy. To nudge only the ones who have not " "acknowledged, use resend; if this second wave is " "deliberate, pass force=true." ), status_code=409, ) async def cancel( self, campaign_id: uuid.UUID, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> SecawCampaign: campaign = await self._require(campaign_id) if campaign.status not in (CAMPAIGN_DRAFT, CAMPAIGN_SCHEDULED): raise AppException( code="SECAW_CAMPAIGN_ALREADY_SENT", message=( "this campaign has already reached people; close it " "instead of cancelling" ), status_code=409, ) campaign.status = CAMPAIGN_CANCELLED campaign.closed_at = datetime.now(UTC) campaign.updated_by_person_id = actor_id await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.campaign.cancel", status="success", resource_type="secaw_campaign", resource_id=str(campaign.id), detail={}, ) return campaign async def close( self, campaign_id: uuid.UUID, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> SecawCampaign: """Declare a sent campaign finished. This stops it appearing as outstanding work for the people who never acknowledged — which is a real decision with a real cost, so it is an explicit act by a named person and it is audited. The unacknowledged rows stay exactly as they are: the coverage report's job is to say 87%, not to round it up. """ campaign = await self._require(campaign_id) if campaign.status != CAMPAIGN_SENT: raise AppException( code="SECAW_CAMPAIGN_NOT_SENT", message=f"cannot close a campaign in status {campaign.status!r}", status_code=409, ) funnel = (await self._funnel_map([campaign.id])).get( campaign.id, self._empty_funnel(), ) campaign.status = CAMPAIGN_CLOSED campaign.closed_at = datetime.now(UTC) campaign.updated_by_person_id = actor_id await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.campaign.close", status="success", resource_type="secaw_campaign", resource_id=str(campaign.id), # The funnel AT CLOSE TIME, in the audit row: closing is when # the number stops moving, so this is the number somebody may # later report. detail={"funnel_at_close": funnel}, ) return campaign async def delete( self, campaign_id: uuid.UUID, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> None: """Only a campaign that never reached anyone. Same rule as materials, same reason: the target ledger is evidence.""" campaign = await self._require(campaign_id) reached = (await self.db.execute( select(func.count()).select_from(SecawCampaignTarget) .where(SecawCampaignTarget.campaign_id == campaign_id) )).scalar() or 0 if reached: raise AppException( code="SECAW_CAMPAIGN_HAS_TARGETS", message=( "this campaign has already been sent to people; close " "it instead of deleting" ), status_code=409, ) snapshot = self._snapshot(campaign) await self.db.delete(campaign) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.campaign.delete", status="success", resource_type="secaw_campaign", resource_id=str(campaign_id), detail=snapshot, ) # ════════════════════════════════════════════════════════════════ # Dispatch # ════════════════════════════════════════════════════════════════ async def dispatch_due( self, *, trace_id: str | uuid.UUID = "", limit: int = 20, ) -> dict[str, Any]: """Cron entry point: send every campaign whose moment has come. One resolver for the whole tick — the person pool and the org tree are the expensive part and they do not change between two campaigns firing in the same minute. """ now = datetime.now(UTC) # Re-arm dead runs first, so the loop below picks them up in the # same tick. Phase 1 skips everyone already queued, so a resumed # dispatch reaches exactly the people the dead one never got to — # nobody gets a second card out of this. recovered = await self.db.execute( update(SecawCampaign) .where( SecawCampaign.status == CAMPAIGN_SENDING, SecawCampaign.updated_at < now - STALE_SENDING_AFTER, ) .values(status=CAMPAIGN_SCHEDULED, updated_at=now) ) stale_recovered = getattr(recovered, "rowcount", 0) or 0 if stale_recovered: await self.db.commit() logger.warning( "secaw.dispatch_due: re-armed %s campaign(s) wedged in " "'sending' for over %s", stale_recovered, STALE_SENDING_AFTER, ) due_ids = (await self.db.execute( select(SecawCampaign.id) .where( SecawCampaign.status == CAMPAIGN_SCHEDULED, SecawCampaign.scheduled_at.is_not(None), SecawCampaign.scheduled_at <= now, ) .order_by(SecawCampaign.scheduled_at.asc()) .limit(limit) )).scalars().all() resolver = AudienceResolver(self.db) results: list[dict[str, Any]] = [] for campaign_id in due_ids: try: results.append(await self.dispatch( campaign_id, trace_id=trace_id, resolver=resolver, )) except Exception: # noqa: BLE001 # One campaign's failure must not stop the others. The # failure is recorded on the row (``last_dispatch_error``) # by ``dispatch`` itself before it re-raises, so the # console shows it even though this handler swallows it. logger.exception( "secaw.dispatch.failed campaign=%s", campaign_id, ) results.append({ "campaign_id": str(campaign_id), "error": True, }) return { "considered": len(due_ids), "dispatched": sum(1 for r in results if not r.get("error")), "failed": sum(1 for r in results if r.get("error")), "stale_recovered": stale_recovered, "results": results, } async def dispatch( self, campaign_id: uuid.UUID, *, trace_id: str | uuid.UUID = "", resolver: AudienceResolver | None = None, resend_only: bool = False, actor_id: uuid.UUID | None = None, ) -> dict[str, Any]: """Send (or resend) one campaign. Phases, each committed before the next — see the module docstring for why a single transaction across the fan-out is not an option. """ campaign = await self._require(campaign_id) if resend_only: if campaign.status != CAMPAIGN_SENT: raise AppException( code="SECAW_CAMPAIGN_NOT_SENT", message="only a sent campaign can be resent", status_code=409, ) else: # Phase 0 — claim it. A conditional UPDATE, so a second # dispatch tick on the same row changes nothing and returns. claimed = await self.db.execute( update(SecawCampaign) .where( SecawCampaign.id == campaign_id, SecawCampaign.status == CAMPAIGN_SCHEDULED, ) .values(status=CAMPAIGN_SENDING, updated_at=datetime.now(UTC)) ) await self.db.commit() # ``rowcount`` is on ``CursorResult``, which is what an UPDATE # actually returns; the declared ``Result`` union does not # carry it. Zero rows = another tick already claimed this # campaign, which is the whole point of the conditional. if (getattr(claimed, "rowcount", 0) or 0) == 0: return { "campaign_id": str(campaign_id), "skipped": "not_scheduled", } campaign = await self._require(campaign_id) try: return await self._dispatch_inner( campaign, trace_id=trace_id, resolver=resolver, resend_only=resend_only, actor_id=actor_id, ) except BaseException as exc: # noqa: BLE001 # ``BaseException``, not ``Exception``: the failure this # compensation exists for is a CANCELLED dispatch — a deploy # replacing the container, or the client hanging up on the # in-request send — and ``asyncio.CancelledError`` is a # ``BaseException`` since 3.8. Catching only ``Exception`` # meant the one case that actually happened in production # skipped the compensation entirely and left the campaign # wedged in ``sending`` with no recorded reason (2026-09-07). # Re-raised unchanged, so cancellation still cancels. # # savepoint-exempt: the block opens with a ``rollback()``, so it # runs on a freshly reactivated transaction rather than inside the # caller's failed one, and it contains only two Core UPDATEs and a # commit — no ORM flush that a savepoint could isolate. On failure # it rolls back again (so ``dispatch_due`` can move on to the next # campaign on the same session) and the ORIGINAL exception is # re-raised either way; nothing is swallowed. try: await self.db.rollback() # Un-arm ONLY a campaign still sitting in ``sending`` — # i.e. one that died before phase 2 settled the status. # Once it is ``sent``, people hold cards and their outbox # rows are durable: putting it back to ``scheduled`` # would have the next tick re-enqueue the whole audience # and send every one of them a second copy. (For a # resend the status is ``sent`` throughout, so this # conditional is also what keeps a failed 催办 from # un-sending the campaign it was chasing.) await self.db.execute( update(SecawCampaign) .where( SecawCampaign.id == campaign_id, SecawCampaign.status == CAMPAIGN_SENDING, ) .values( status=CAMPAIGN_SCHEDULED, updated_at=datetime.now(UTC), ) ) # Record WHY regardless of status. A green cron run and a # silent non-send look identical from the console # otherwise. await self.db.execute( update(SecawCampaign) .where(SecawCampaign.id == campaign_id) # ``str(CancelledError())`` is the EMPTY STRING, so a # cancelled dispatch would have recorded a reason of # "" — indistinguishable from a clean run in the # console. Fall back to the exception's class name. .values( last_dispatch_error=( str(exc)[:500] or type(exc).__name__ ), ) ) await self.db.commit() except Exception: # noqa: BLE001 # Never let the bookkeeping replace the real failure — # especially not during a shutdown, where the session may # already be unusable. logger.exception( "secaw.dispatch: could not record the failure for " "campaign %s", campaign_id, ) # Hand a usable session back: ``dispatch_due`` swallows one # campaign's failure and dispatches the NEXT one on this very # session, so a pending-rollback left here would turn one # wedged campaign into every remaining campaign failing. with contextlib.suppress(Exception): await self.db.rollback() raise async def _dispatch_inner( self, campaign: SecawCampaign, *, trace_id: str | uuid.UUID, resolver: AudienceResolver | None, resend_only: bool, actor_id: uuid.UUID | None, ) -> dict[str, Any]: material = await self.db.get(SecawMaterial, campaign.material_id) if material is None: # pragma: no cover - FK guarantees it raise AppException( code="SECAW_MATERIAL_MISSING", message="the campaign's material no longer exists", status_code=409, ) # Phase 1 — resolve the audience and write the target ledger. if resend_only: members = await self._members_for_resend(campaign) else: resolver = resolver or AudienceResolver(self.db) members = await resolver.resolve(campaign.audience_config) if len(members) > MAX_AUDIENCE: raise AppException( code="SECAW_AUDIENCE_OVER_CAP", message=( f"audience resolves to {len(members)} people, over " f"the {MAX_AUDIENCE} cap" ), status_code=409, ) version = material.version now = datetime.now(UTC) existing = { t.person_id: t for t in (await self.db.execute( select(SecawCampaignTarget).where( SecawCampaignTarget.campaign_id == campaign.id, ) )).scalars().all() } already_queued: set[uuid.UUID] = set() for member in members: target = existing.get(member.person.id) if target is None: target = SecawCampaignTarget( id=uuid.uuid4(), campaign_id=campaign.id, person_id=member.person.id, person_snapshot=member.snapshot, material_version=version, lang=member.lang, created_at=now, ) self.db.add(target) elif resend_only: # A resend does not rewrite the snapshot: what they were # sent the first time is part of the record. target.resend_count += 1 elif target.queued_at is not None: # A FIRST send that finds this person already queued is a # resumed dispatch (the previous attempt died between two # phase-1 commits, or a deploy cancelled it). They already # have a durable outbox row — the retry worker owns it — # so enqueueing a second one is how 400 people get the # same card twice. Counted, not re-sent. NOT a # ``resend_count`` bump either: nobody chased them. already_queued.add(member.person.id) if not resend_only: campaign.material_version = version await self.db.commit() # Phase 2 — settle the campaign's status BEFORE the fan-out. # # The ledger is now complete and durable, which is the whole of # what ``sent`` claims (per-channel delivery truth lives in the # outbox and always has — ``SecawCampaignTarget`` has no # ``delivered_at`` on purpose). Leaving the flip until after the # drain made the status a *lie in the other direction* for as # long as the drain took: every employee-facing read gates on # ``status == 'sent'`` (``my_pending``, # ``pending_all_for_material``, ``mark_opened`` via # ``CAMPAIGN_LIVE_STATUSES``, ``acknowledge``, the reminder # ladder, the compliance report), so during a 416-person drain # the people whose card had ALREADY arrived opened the material # and found no confirm button — the ack endpoint answers 409 # ``SECAW_CAMPAIGN_NOT_OPEN`` while the campaign is ``sending``. # Measured on the 2026-09-07 send: a 15.5-minute window in which # 264 people held a card asking them to confirm something the # platform would not let them confirm, and their ``opened_at`` # was dropped on the floor. Phase 1 takes ~20s for the same # audience. if not resend_only: campaign.status = CAMPAIGN_SENT campaign.sent_at = now await self.audit.log( trace_id=trace_id, actor_type="user" if actor_id else "system", actor_user_id=actor_id, action=( "security_awareness.campaign.resend" if resend_only else "security_awareness.campaign.send" ), status="success", resource_type="secaw_campaign", resource_id=str(campaign.id), detail={ "audience": len(members), "material_version": version, "already_queued": len(already_queued), "to_send": len(members) - len(already_queued), }, ) await self.db.commit() # Phase 3 — the fan-out. Commits per recipient; never holds a # transaction across the external calls. ``queued_at`` / # ``last_sent_at`` are stamped per recipient INSIDE the enqueue # transaction (``on_enqueued``), not in one bulk UPDATE at the # end: the bulk write meant a drain that died at 90% left all 416 # rows unstamped, so the funnel read 0 queued and a resumed # dispatch could not tell who had already been reached. to_send = [ m for m in members if m.person.id not in already_queued ] envelopes = [ self._envelope(campaign, material, member) for member in to_send ] campaign_id = campaign.id async def _stamp_queued(env: Any, _outbox_id: uuid.UUID) -> None: """Mark this recipient's ledger row as handed to the outbox. Core UPDATE keyed on ids, never an ORM instance: phase 1's error path rolls back, which expires every instance in the identity map, and touching one after that is a lazy load ⇒ ``MissingGreenlet`` out of a helper that promises never to raise for one recipient. """ await self.db.execute( update(SecawCampaignTarget) .where( SecawCampaignTarget.campaign_id == campaign_id, SecawCampaignTarget.person_id == env.recipient.id, ) .values(queued_at=now, last_sent_at=now) ) from app.notifications import fanout result = await fanout.enqueue_then_drain( self.db, envelopes, trace_id=str(trace_id or ""), actor_user_id=actor_id, label="security_awareness.campaign", on_enqueued=_stamp_queued, ) # Phase 4 — record whether the fan-out had losses. The status is # already settled; this only annotates it. fresh = await self._require(campaign_id) fresh.last_dispatch_error = ( f"enqueue failed for {result['failed']} recipient(s)" if result["failed"] else None ) await self.db.commit() return { "campaign_id": str(campaign_id), "audience": len(members), "already_queued": len(already_queued), **result, } async def require_resendable(self, campaign_id: uuid.UUID) -> None: """409 unless this campaign can be resent. Split out so the HTTP layer can answer the admin synchronously while the fan-out itself runs off-request: a 202 followed by a background refusal is a click that silently did nothing. """ campaign = await self._require(campaign_id) if campaign.status != CAMPAIGN_SENT: raise AppException( code="SECAW_CAMPAIGN_NOT_SENT", message="only a sent campaign can be resent", status_code=409, ) async def resend_pending( self, campaign_id: uuid.UUID, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> dict[str, Any]: """Nudge only the people who have not acknowledged. Deliberately not "resend to everybody": re-pushing to somebody who already did what was asked is the fastest way to teach the whole company to ignore this channel. """ return await self.dispatch( campaign_id, trace_id=trace_id, resend_only=True, actor_id=actor_id, ) async def _members_for_resend( self, campaign: SecawCampaign, ) -> list[AudienceMember]: """Rebuild ``AudienceMember`` for the unacknowledged targets. From the STORED target rows, not by re-resolving the audience: a resend must reach the people this campaign was actually sent to, not whoever the filter would select today. """ rows = (await self.db.execute( select(SecawCampaignTarget, IdentityPerson) .join(IdentityPerson, IdentityPerson.id == SecawCampaignTarget.person_id) .where( SecawCampaignTarget.campaign_id == campaign.id, SecawCampaignTarget.acknowledged_at.is_(None), # A person who has since left is skipped here rather # than left to the outbox's leaver gate: the gate would # mark the row dead, which is correct but shows up in the # console as a delivery failure rather than as a # deliberate skip. IdentityPerson.employment_status != "left", IdentityPerson.deleted_at.is_(None), ) )).all() return [ AudienceMember( person=person, lang=target.lang, snapshot=target.person_snapshot or {}, ) for target, person in rows ] def _envelope( self, campaign: SecawCampaign, material: SecawMaterial, member: AudienceMember, ) -> Any: """One recipient's card / mail / bell row. The material's title and summary are resolved into the recipient's own language HERE, using the same ``service.pick_block`` the web reader uses — so the card, the mail, the bell and the page they land on all say the same thing in the same language. """ from app.notifications.service import ( _SEE_TEMPLATE_SENTINEL, CardStat, LocalizedText, NotificationEnvelope, ) lang, block, _fb = pick_block(material.localized, member.lang) title = str(block.get("title") or "") summary = block.get("summary") minutes = estimate_minutes( material.kind, block, material.estimated_minutes, ) url = MATERIAL_ROUTE.format(material_id=material.id) # Both are pre-rendered per recipient because they are optional # fragments inside a localized sentence — a template placeholder # cannot be conditionally absent. minutes_label = { "zh": f"(约 {minutes} 分钟)", "en": f" (~{minutes} min)", "ja": f"(約 {minutes} 分)", }.get(lang, "") if minutes else "" summary_line = f"{summary}\n\n" if summary else "" due_label = self._due_label(campaign.due_at, lang) return NotificationEnvelope( recipient=member.person, title=title, # Sentinel rather than rendered Chinese: ``render_inbox_body`` # returns a stored non-sentinel body verbatim, so the bell # would keep showing Chinese to a reader whose card and mail # already render the localised template. body_text=_SEE_TEMPLATE_SENTINEL, template_key=( TEMPLATE_MANDATORY if campaign.mandatory else TEMPLATE_NOTICE ), channels=list(campaign.channels or []), url=url, context={ "material_title": title, "minutes_label": minutes_label, "summary_line": summary_line, "due_label": due_label, }, related_resource_type=CAMPAIGN_RESOURCE_TYPE, related_resource_id=campaign.id, card_subtitle=LocalizedText( zh="安全宣贯 · 必读" if campaign.mandatory else "安全宣贯", en=( "Security awareness · required" if campaign.mandatory else "Security awareness" ), ja=( "セキュリティ啓発 · 必読" if campaign.mandatory else "セキュリティ啓発" ), ), card_stats=( [ CardStat( label=LocalizedText(zh="截止", en="Due", ja="期限"), value=LocalizedText( zh=due_label, en=due_label, ja=due_label, ), ), ] if campaign.mandatory and campaign.due_at else [] ), ) @staticmethod def _due_label(due_at: datetime | None, lang: str) -> str: if due_at is None: # An empty due date must not render as "None" inside a # sentence. "尽快" is the honest reading of a mandatory # campaign without a deadline. return {"zh": "尽快", "en": "as soon as you can", "ja": "お早めに"}.get( lang, "尽快", ) return due_at.strftime("%Y-%m-%d") # ════════════════════════════════════════════════════════════════ # Internals # ════════════════════════════════════════════════════════════════ async def _require(self, campaign_id: uuid.UUID) -> SecawCampaign: row = await self.db.get(SecawCampaign, campaign_id) if row is None: raise NotFoundError("security awareness campaign", str(campaign_id)) return row @staticmethod def _clean_channels(channels: list[str] | None) -> list[str]: """Keep only known channels, in a stable order. An empty list is legal and means "bell only" — the in-app inbox row is written by every dispatch regardless, so a campaign with no push channels is a quiet one, not a broken one. """ wanted = set(channels or []) return [c for c in CAMPAIGN_CHANNELS if c in wanted] @staticmethod def _empty_funnel() -> dict[str, int]: return { "targeted": 0, "queued": 0, "delivered": 0, "opened": 0, "acknowledged": 0, } async def _funnel_map( self, campaign_ids: list[uuid.UUID], ) -> dict[uuid.UUID, dict[str, int]]: """``{campaign_id: {targeted, queued, delivered, opened, acknowledged}}``. ``delivered`` comes from the outbox — the only place that knows. The other four come from the target ledger. """ if not campaign_ids: return {} rows = (await self.db.execute( select( SecawCampaignTarget.campaign_id, func.count(), func.count(SecawCampaignTarget.queued_at), func.count(SecawCampaignTarget.opened_at), func.count(SecawCampaignTarget.acknowledged_at), ) .where(SecawCampaignTarget.campaign_id.in_(campaign_ids)) .group_by(SecawCampaignTarget.campaign_id) )).all() out: dict[uuid.UUID, dict[str, int]] = { cid: { "targeted": int(targeted), "queued": int(queued), "delivered": 0, "opened": int(opened), "acknowledged": int(acked), } for cid, targeted, queued, opened, acked in rows } # ``related_resource_id`` is a bare ``CHAR(36)``, NOT the GUID # type — it stores the hyphenated string, so a raw ``uuid.UUID`` # on the other side of the comparison matches nothing and the # count comes back 0 while looking like it worked. Every other # caller in this codebase wraps in ``str()`` for the same reason. id_strings = [str(c) for c in campaign_ids] delivered = (await self.db.execute( select( NotificationOutbox.related_resource_id, func.count(func.distinct(NotificationOutbox.recipient_person_id)), ) .where( NotificationOutbox.related_resource_type == CAMPAIGN_RESOURCE_TYPE, NotificationOutbox.related_resource_id.in_(id_strings), # The constants, not a literal: the outbox's success # status is ``delivered``, and a hardcoded "sent" here # matched nothing — the funnel would have reported 0 # delivered forever while looking like it worked. # ``partial`` counts too: a card that reached Feishu but # whose email bounced DID reach the person. NotificationOutbox.status.in_(_DELIVERED_STATUSES), ) .group_by(NotificationOutbox.related_resource_id) )).all() # The grouped key comes back as the stored STRING; the map is # keyed by UUID. by_str = {str(cid): cid for cid in campaign_ids} for cid_str, count in delivered: cid = by_str.get(str(cid_str)) if cid is not None and cid in out: out[cid]["delivered"] = int(count) return out async def _delivery_map( self, campaign_id: uuid.UUID, person_ids: list[uuid.UUID], ) -> dict[uuid.UUID, dict[str, Any]]: """Per-person delivery outcome, straight from the outbox. Returns ``status`` plus the two fields that turn a status into an explanation: which channels failed, and the error they failed with. ``partial`` on its own is not actionable — it is equally the shape of "Feishu is not configured for this person" and "Feishu returned a one-off 500", and those want opposite responses from the admin. When a person has several outbox rows for one campaign (a resend makes a second one), the LATEST row wins — that is the attempt whose outcome the admin is asking about. """ if not person_ids: return {} rows = (await self.db.execute( select( NotificationOutbox.recipient_person_id, NotificationOutbox.status, NotificationOutbox.failed_channels, NotificationOutbox.last_error, NotificationOutbox.created_at, ) .where( NotificationOutbox.related_resource_type == CAMPAIGN_RESOURCE_TYPE, # ``str()`` — see the note in ``_funnel_map``. NotificationOutbox.related_resource_id == str(campaign_id), NotificationOutbox.recipient_person_id.in_(person_ids), ) .order_by(NotificationOutbox.created_at.asc()) )).all() out: dict[uuid.UUID, dict[str, Any]] = {} for person_id, status, failed_channels, last_error, _created in rows: out[person_id] = { "status": status, "failed_channels": list(failed_channels or []), # Truncated: the column holds up to 4000 chars of joined # per-channel errors and this is a table cell, not a log # viewer. The full text stays one click away in the # notification outbox, which is where forensics belongs. "error": (last_error or None) and str(last_error)[:300], } return out async def _breakdown( self, campaign_id: uuid.UUID, dimension: str, ) -> list[dict[str, Any]]: """Acknowledgement rate per department / country. Grouped on the SNAPSHOT, not on the person's current placement: somebody who transferred out last month was reached as a member of their old team, and moving their row would silently rewrite a past department's coverage number. """ rows = (await self.db.execute( select( SecawCampaignTarget.person_snapshot, SecawCampaignTarget.acknowledged_at, SecawCampaignTarget.opened_at, ).where(SecawCampaignTarget.campaign_id == campaign_id) )).all() buckets: dict[str, dict[str, int]] = {} for snapshot, acked, opened in rows: key = str((snapshot or {}).get(dimension) or "—") b = buckets.setdefault(key, {"targeted": 0, "opened": 0, "acknowledged": 0}) b["targeted"] += 1 if opened: b["opened"] += 1 if acked: b["acknowledged"] += 1 return [ {"key": k, **v} for k, v in sorted( buckets.items(), key=lambda kv: (-kv[1]["targeted"], kv[0]), ) ] async def _material_titles( self, material_ids: set[uuid.UUID], ) -> dict[uuid.UUID, str | None]: """Baseline (zh) titles, for the ADMIN console only. The admin console is authored-in-Chinese by the security desk; employee-facing surfaces resolve the language per reader through ``SecurityAwarenessService``. """ ids = [i for i in material_ids if i] if not ids: return {} rows = (await self.db.execute( select(SecawMaterial.id, SecawMaterial.localized) .where(SecawMaterial.id.in_(ids)) )).all() out: dict[uuid.UUID, str | None] = {} for mid, localized in rows: blocks = localized or {} for lang in ("zh", "en", "ja"): block = blocks.get(lang) if isinstance(block, dict) and block.get("title"): out[mid] = block["title"] break else: out[mid] = None return out @staticmethod def _campaign_row(c: SecawCampaign) -> dict[str, Any]: return { "id": str(c.id), "material_id": str(c.material_id), "material_version": c.material_version, "name": c.name, "audience_config": c.audience_config or {}, "channels": list(c.channels or []), "mandatory": c.mandatory, "completion_mode": c.completion_mode, "status": c.status, "scheduled_at": c.scheduled_at, "due_at": c.due_at, "sent_at": c.sent_at, "closed_at": c.closed_at, "last_dispatch_error": c.last_dispatch_error, "created_at": c.created_at, "updated_at": c.updated_at, } @staticmethod def _snapshot(c: SecawCampaign) -> dict[str, Any]: return { "name": c.name, "material_id": str(c.material_id), "mandatory": c.mandatory, "completion_mode": c.completion_mode, "channels": list(c.channels or []), "status": c.status, "audience_config": c.audience_config or {}, "due_at": c.due_at.isoformat() if c.due_at else None, } # NOTE — there is deliberately no "record a campaign_card impression" # writer here any more. It had no caller, and the only way to reach it # would have been to trust a browser's word that it had seen a push card. # Whether a campaign message reached somebody is the outbox's # ``delivered``/``partial`` status — the thing that owns the retry ladder # and the leaver gate — and ``_funnel_map`` reads it from there. Writing # a second copy of that fact into the exposure ledger is exactly the # mirror this module's docstring refuses to keep. ``SURFACE_CAMPAIGN_CARD`` # and ``SecawImpression.campaign_id`` stay defined: historical rows may # carry them, and an inbound Feishu card-open callback (which would be # real evidence, unlike a self-report) is the shape that would use them. async def run_campaign_dispatch_in_background( campaign_id: uuid.UUID, *, trace_id: str, actor_id: uuid.UUID | None = None, resend_only: bool = False, ) -> None: """Dispatch one campaign on a FRESH session, off the request thread. The coroutine ``/send-now`` hands to ``background.spawn`` after committing the (fast, all-DB) arming step and returning to the admin. Same shape as ``asset.inventory_service. run_start_notifications_in_background``, for the same reason: a company-wide fan-out is hundreds of external calls — 15.5 minutes for 416 people × (Feishu card + email), measured 2026-09-07 — and running it inside the request meant * nginx gave up at its 300s ``proxy_read_timeout`` and the drawer showed a generic failure over a send that was still in flight, which is the single worst thing to show an admin holding a button that reaches 416 people, and * the admin's next move (reopen, press again) created a SECOND campaign rather than retrying the first. Owns its own session, because the request's is closed the moment the response goes out. Best-effort by construction: on a crash the outbox rows already enqueued are durably PENDING for the ``notification.outbox_retry`` worker, and the campaign is either already ``sent`` (phase 2 settles the status before the drain) or re-armed by the compensation in ``dispatch`` / the stale-``sending`` sweep in ``dispatch_due``. The admin's click is therefore never lost even if this coroutine never gets to run at all — the campaign is ``scheduled`` before it is spawned, and the ten-minute ``security_awareness.campaign_dispatch`` tick is the floor under it. """ from sqlalchemy import text as _text from app.core.database import async_session_factory try: async with async_session_factory() as session: # A 2000-person ceiling of chained ``notification_outbox`` # inserts, each taking a shared FK lock on a person row — # same reason the inventory fan-out and the HRIS # background-apply raise this. await session.execute( _text("SET SESSION innodb_lock_wait_timeout = 120") ) result = await CampaignService(session).dispatch( campaign_id, trace_id=trace_id, actor_id=actor_id, resend_only=resend_only, ) logger.info( "secaw.send_now background dispatch finished for %s: %s", campaign_id, result, ) except Exception: # noqa: BLE001 logger.exception( "secaw.send_now background dispatch crashed for campaign %s — " "rows enqueued so far stay PENDING for notification." "outbox_retry, and the campaign is re-armed for the next " "campaign_dispatch tick.", campaign_id, ) __all__ = [ "ADMIN_MAX_PAGE_SIZE", "COMPLETION_ACK", "COMPLETION_IMPRESSION", "COMPLETION_OPEN", "DUPLICATE_SEND_WINDOW", "STALE_SENDING_AFTER", "TARGET_MAX_PAGE_SIZE", "CampaignService", "run_campaign_dispatch_in_background", ]