"""CRUD service for security tips. Every mutation: * Validates input lengths (empty / over-500 raises ``AppException``). * Updates ``display_order`` on create so newly added tips land at the end of the list (admins can drag-reorder afterwards). * Logs to ``audit_logs`` with action ``platform.security_tip.``. Public read (``list_visible``) is exposed without an admin gate — any signed-in user fetches the enabled set for the header banner. """ from __future__ import annotations import uuid from datetime import UTC, datetime from sqlalchemy import asc, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.domains.platform_tips.models import PlatformSecurityTip from app.domains.platform_tips.schemas import ( SecurityTipCreate, SecurityTipUpdate, ) class PlatformTipsService: def __init__(self, db: AsyncSession): self.db = db self.audit = AuditService(db) # ── Public read ───────────────────────────────────────────── async def list_visible(self) -> list[PlatformSecurityTip]: """Enabled tips ordered by ``display_order`` ASC, with a stable ``created_at`` tiebreak so concurrent writes don't shuffle the front-end's cycle on every poll.""" rows = (await self.db.execute( select(PlatformSecurityTip) .where(PlatformSecurityTip.enabled.is_(True)) .order_by( asc(PlatformSecurityTip.display_order), asc(PlatformSecurityTip.created_at), ) )).scalars().all() return list(rows) # ── Admin read ────────────────────────────────────────────── async def list_all(self) -> list[PlatformSecurityTip]: """All tips (enabled + disabled) for the admin management page, ordered the same way ``list_visible`` is — keeps preview parity with the live banner.""" rows = (await self.db.execute( select(PlatformSecurityTip).order_by( asc(PlatformSecurityTip.display_order), asc(PlatformSecurityTip.created_at), ) )).scalars().all() return list(rows) # ── Admin mutations ──────────────────────────────────────── async def create( self, body: SecurityTipCreate, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> PlatformSecurityTip: # Place the new tip at the end so existing rotation isn't # disturbed. ``MAX(display_order) + 1`` over an empty table is # NULL → coerce to -1 so the new row lands at 0. max_order = (await self.db.execute( select(PlatformSecurityTip.display_order) .order_by(PlatformSecurityTip.display_order.desc()) .limit(1) )).scalar() new_order = (max_order if max_order is not None else -1) + 1 now = datetime.now(UTC) row = PlatformSecurityTip( id=uuid.uuid4(), text_zh=body.text_zh.strip(), text_en=body.text_en.strip(), # Optional third language (migration 598). Empty-after-trim # collapses to NULL so " " never masquerades as a # translation. text_ja=(body.text_ja or "").strip() or None, enabled=body.enabled, display_order=new_order, 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="platform.security_tip.create", status="success", resource_type="platform_security_tip", resource_id=str(row.id), detail={ "text_zh": row.text_zh, "text_en": row.text_en, "text_ja": row.text_ja, "enabled": row.enabled, }, ) return row async def update( self, tip_id: uuid.UUID, body: SecurityTipUpdate, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> PlatformSecurityTip: row = await self.db.get(PlatformSecurityTip, tip_id) if row is None: raise NotFoundError(f"security tip {tip_id} not found") before = { "text_zh": row.text_zh, "text_en": row.text_en, "text_ja": row.text_ja, "enabled": row.enabled, } changed = False if body.text_zh is not None: cleaned = body.text_zh.strip() if not cleaned: raise AppException( code="SECURITY_TIP_TEXT_EMPTY", message="text_zh cannot be empty after trimming", status_code=400, ) if cleaned != row.text_zh: row.text_zh = cleaned changed = True if body.text_en is not None: cleaned = body.text_en.strip() if not cleaned: raise AppException( code="SECURITY_TIP_TEXT_EMPTY", message="text_en cannot be empty after trimming", status_code=400, ) if cleaned != row.text_en: row.text_en = cleaned changed = True if body.text_ja is not None: # Unlike zh / en this one MAY be cleared — an admin removing a # bad machine translation is a legitimate edit, and the # renderer falls back rather than showing a blank. cleaned_ja = body.text_ja.strip() or None if cleaned_ja != row.text_ja: row.text_ja = cleaned_ja changed = True if body.enabled is not None and body.enabled != row.enabled: row.enabled = body.enabled changed = True if not changed: # No-op update — return the row unchanged. Avoid spamming # the audit log with empty diffs but DON'T raise; the admin # UI may save without changes after a network retry. return row row.updated_at = datetime.now(UTC) row.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="platform.security_tip.update", status="success", resource_type="platform_security_tip", resource_id=str(row.id), detail={ "before": before, "after": { "text_zh": row.text_zh, "text_en": row.text_en, "enabled": row.enabled, }, }, ) return row async def delete( self, tip_id: uuid.UUID, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> None: row = await self.db.get(PlatformSecurityTip, tip_id) if row is None: raise NotFoundError(f"security tip {tip_id} not found") snapshot = { "text_zh": row.text_zh, "text_en": row.text_en, "text_ja": row.text_ja, "enabled": row.enabled, "display_order": row.display_order, } await self.db.delete(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="platform.security_tip.delete", status="success", resource_type="platform_security_tip", resource_id=str(tip_id), detail=snapshot, ) async def reorder( self, ordered_ids: list[uuid.UUID], *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> list[PlatformSecurityTip]: """Apply a new display order. Ids in ``ordered_ids`` get positions 0..n-1; rows not in the list are pushed to the end in their existing relative sequence so a stale client (that missed a concurrent insert) doesn't accidentally clobber the new row's order. """ if not ordered_ids: raise AppException( code="SECURITY_TIP_REORDER_EMPTY", message="ordered_ids must not be empty", status_code=400, ) # Reject duplicates — ambiguous intent and would leave one of # the rows at the wrong position depending on dict iteration. if len(set(ordered_ids)) != len(ordered_ids): raise AppException( code="SECURITY_TIP_REORDER_DUPLICATE", message="ordered_ids must be unique", status_code=400, ) # #1801: lock the rows for the read-modify-write so two admins # reordering concurrently can't lost-update each other's # display_order (the banner carousel then shows a duplicated / # contradictory order). Mirrors policy_notices' reorder, which # already carries this guard for the identical bug class. existing = (await self.db.execute( select(PlatformSecurityTip).with_for_update() )).scalars().all() by_id = {r.id: r for r in existing} # Verify every id in the list refers to a real row. Silently # skipping unknown ids would mask client/server desync bugs. unknown = [i for i in ordered_ids if i not in by_id] if unknown: raise AppException( code="SECURITY_TIP_REORDER_UNKNOWN_ID", message=f"unknown tip ids in reorder: {unknown}", status_code=400, ) # Apply positions to the supplied subset. for idx, tip_id in enumerate(ordered_ids): by_id[tip_id].display_order = idx # Push any rows the client didn't include to the end, in their # existing relative order. tail_start = len(ordered_ids) leftover = sorted( (r for r in existing if r.id not in set(ordered_ids)), key=lambda r: (r.display_order, r.created_at), ) for offset, row in enumerate(leftover): row.display_order = tail_start + offset now = datetime.now(UTC) for r in existing: r.updated_at = now r.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="platform.security_tip.reorder", status="success", resource_type="platform_security_tip", resource_id=None, detail={"ordered_ids": [str(i) for i in ordered_ids]}, ) return await self.list_all()