"""Feed domain — service layer. All business rules live here; the router is a thin HTTP adapter. Visibility model: * ``public`` posts are visible to every logged-in employee. * ``department`` posts are visible to (a) the author, (b) anyone whose primary org_membership.org_node_name matches the snapshot saved on the post, (c) IT / HR admins. The department string is snapshotted at post-time so re-orgs don't retroactively hide old content. * System posts (birthday / anniversary / aggregate banners) are always public. Soft-delete: * Posts, comments, announcements carry ``deleted_at``. The list / detail / search endpoints filter out non-NULL rows uniformly. Idempotency: * ``feed_likes`` UNIQUE(post_id, person_id) so a double-click is a no-op. * ``feed_blessings`` UNIQUE(celebration_id, sender_person_id) so a duplicate send is rejected with a 409. The router translates this to a friendly toast. Mentions: * Parsed by the caller (``mentioned_person_ids``). Stored verbatim in ``feed_mentions`` and dispatched to NotificationService. """ from __future__ import annotations import logging import uuid from datetime import UTC, date, datetime, timedelta from typing import Any, cast from sqlalchemy import and_, delete, desc, false, func, or_, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ( ConflictError, ForbiddenError, NotFoundError, ValidationError, ) from app.core.rbac import GLOBAL_ADMIN_ROLE, HR_ROLE, has_role from app.domains.feed.models import ( ANNOUNCEMENT_STATUS_ARCHIVED, ANNOUNCEMENT_STATUS_PUBLISHED, ANONYMITY_ANONYMOUS, ANONYMITY_NAMED, BIRTHDAY_VISIBILITY_HIDDEN, DEFAULT_PREFERENCES, POST_CATEGORY_GENERAL, POST_TYPE_AGGREGATE_ANNIVERSARY, POST_TYPE_AGGREGATE_BIRTHDAY, POST_TYPE_SYSTEM_ANNIVERSARY, POST_TYPE_SYSTEM_BIRTHDAY, POST_TYPE_USER, PREF_BIRTHDAY_VISIBILITY, PREF_FESTIVE_EFFECTS, PREF_THEME_MODE, PREF_WRAPPED_PUSH_OPT_OUT, SYSTEM_POST_TYPES, VISIBILITY_DEPARTMENT, VISIBILITY_PUBLIC, FeedAnnouncement, FeedAttachment, FeedBlessing, FeedCelebration, FeedComment, FeedLike, FeedMention, FeedPost, FeedReport, IdentityPersonPreference, ) from app.identity.person_models import IdentityOrgMembership, IdentityPerson logger = logging.getLogger(__name__) MAX_ATTACHMENTS_PER_POST = 9 MAX_MENTIONS_PER_POST = 20 USER_TAB_MINE = "mine" USER_TAB_DEPARTMENT = "department" USER_TAB_PUBLIC = "public" USER_TABS = frozenset({ USER_TAB_MINE, USER_TAB_DEPARTMENT, USER_TAB_PUBLIC, "anonymous", }) # How far back a celebration may still be blessed. Celebrations are a # same-day social ritual; without a bound, ANY person could open a # years-old celebration by id and send a blessing that fires a fresh # Feishu push at the celebrant ("happy birthday" in October for a March # birthday). One week covers the legitimate late-comer (was on leave / # saw the banner in a weekly digest) and nothing else. BLESSING_WINDOW_DAYS = 7 # Stable error codes for the celebration reply loop. The frontend keys its # localized copy off ``errors[0].code`` — never off the English message. ERR_BLESSING_WINDOW_CLOSED = "BLESSING_WINDOW_CLOSED" ERR_CELEBRATION_ALREADY_THANKED = "CELEBRATION_ALREADY_THANKED" ERR_CELEBRATION_NO_BLESSINGS = "CELEBRATION_NO_BLESSINGS" def _iso_utc(dt: datetime | None) -> str | None: """ISO-format a datetime, anchoring naive values to UTC. MySQL ``DATETIME`` columns (even ``DateTime(timezone=True)``) lose ``tzinfo`` on a fresh load; emitting a naive ISO string makes the browser interpret the timestamp as local time → 8h drift in Asia/Shanghai. Anchoring naive values to UTC at the serialisation boundary is the same fix used by /api/v1/feed announcement / mention paths. """ if dt is None: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=UTC) return dt.astimezone(UTC).isoformat() def _validate_attachment_keys( attachments: list[dict[str, Any]], *, owner_id: Any, ) -> None: """Reject any object key the caller is not entitled to attach. The create path used to store the client-supplied ``s3_key`` verbatim — the only constraint was its length — and hydration then minted a 10-minute presigned GET for whatever was in the column. So any active employee could post ``attachments=[{"s3_key": "auditor_evidence/…/vendor-soc2.pdf"}]`` and read the download URL straight out of the create response (authors always see their own post). That defeats ``/uploads/presign-get``'s per-purpose ACL — the endpoint whose own docstring says bare authentication is not enough because "the presigned URL bypasses Helix auth once minted" — for every namespace sharing the bucket: ``inventory_evidence`` (other people's device photos), ``vendor_assessment``, ``self_check``. Deleting the post afterwards does not help: the attachment row survives and the URL is already in hand (#2124). Two conditions, both from the key path itself, which ``presign_put`` stamps with the uploader's id: * the purpose must be one of the feed's own, and * the uploader must be this author — you may attach YOUR upload. """ from app.storage_router import FEED_PURPOSES, parse_object_key seen_keys: set[str] = set() for att in attachments: key = (att.get("s3_key") or "").strip() if key in seen_keys: raise ValidationError("attachment object key is duplicated") seen_keys.add(key) parsed = parse_object_key(key) if parsed is None: raise ValidationError("attachment object key is not recognised") purpose, uploader_id = parsed if purpose not in FEED_PURPOSES: raise ValidationError( f"attachment purpose '{purpose}' cannot be posted to the feed", ) if owner_id is not None and str(uploader_id) != str(owner_id): raise ValidationError( "attachment was uploaded by someone else", ) def _banner_occasion_date(post: FeedPost) -> date | None: """Which celebration day an aggregate banner represents. Read order: ``body_params["occasion_date"]`` first, ``created_at`` date only as the legacy fallback. ``created_at`` CANNOT be used to derive this. The banner's celebrants are indexed by ``FeedCelebration.occasion_date``, which the celebration cron writes as an **Asia/Shanghai** wall-clock date, while ``created_at`` is stored in **UTC**. The cron creates each day's banner on the first hourly tick of the SHA day, i.e. between 00:00 and 07:59 SHA — which is the PREVIOUS calendar day in UTC. So ``created_at.date()`` was off by one for every banner the cron has ever created, the ``(date, type)`` lookup missed 100 % of the time, and the aggregate card rendered with an empty avatar group and no 查看 button. A banner created at exactly 08:00 SHA or later would have matched — the cron never runs then, which is why this looked like "the feature is just empty" rather than a flaky bug. ``celebration_service`` now stamps the SHA date explicitly into ``body_params``. Banners written before that stay on the old inference, so they keep whatever behaviour they already had instead of raising. """ raw = (post.body_params or {}).get("occasion_date") if isinstance(raw, date) and not isinstance(raw, datetime): return raw if isinstance(raw, str): try: return date.fromisoformat(raw) except ValueError: logger.warning( "feed.banner.bad_occasion_date post=%s raw=%r", post.id, raw, ) return post.created_at.date() if post.created_at is not None else None def _resolve_dept(person: IdentityPerson) -> str | None: return person.primary_department def _is_admin(actor: IdentityPerson) -> bool: """Feed-domain admin = global_admin OR hr. Used to gate announcement CRUD + cross-author post / comment deletion. The role token is ``hr`` (not ``hr_admin``) — the system role table seeds it that way (see ``app.core.rbac.HR_ROLE``) and every other HR-gated route in the codebase checks the same token. Getting this wrong locks HR out of the /admin/announcements page entirely. """ return has_role(actor.role_names, GLOBAL_ADMIN_ROLE, HR_ROLE) class FeedService: def __init__(self, db: AsyncSession) -> None: self.db = db # ──────────────────────────────────────────────────────────── # Person hydration # ──────────────────────────────────────────────────────────── async def _hydrate_persons( self, person_ids: set[str], ) -> dict[str, IdentityPerson]: if not person_ids: return {} rows = (await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_([uuid.UUID(p) for p in person_ids]), ) )).scalars().all() return {str(p.id): p for p in rows} @staticmethod def _person_lite(p: IdentityPerson | None) -> dict[str, Any] | None: if p is None: return None return { "id": str(p.id), "display_name": p.display_name, "work_email": p.work_email, "avatar_url": p.avatar_url, "department": p.primary_department, } @staticmethod def _blessing_sender(p: IdentityPerson | None) -> dict[str, Any] | None: """Sender projection for blessings — no ``work_email``. ``/celebrations/{id}`` fetches any celebration by id for any authenticated user with no per-row visibility check, so echoing the sender's ``work_email`` (as the shared ``_person_lite`` does) turned id-enumeration into a roster-email leak. The blessing card UI only needs name / avatar / department, matching the celebrant projection in the router. Do NOT reuse ``_person_lite`` here. """ if p is None: return None return { "id": str(p.id), "display_name": p.display_name, "avatar_url": p.avatar_url, "department": p.primary_department, } # ──────────────────────────────────────────────────────────── # Visibility # ──────────────────────────────────────────────────────────── def _can_view_post( self, post: FeedPost, viewer: IdentityPerson, ) -> bool: if post.deleted_at is not None: return False if post.post_type in SYSTEM_POST_TYPES: return True if post.visibility == VISIBILITY_PUBLIC: return True if _is_admin(viewer): return True if ( post.author_person_id is not None and str(post.author_person_id) == str(viewer.id) ): return True if ( post.visibility == VISIBILITY_DEPARTMENT and post.department_snapshot and post.department_snapshot == _resolve_dept(viewer) ): return True return False @staticmethod def _is_celebrant_of(post: FeedPost, viewer: IdentityPerson) -> bool: """True when ``post`` is the viewer's OWN celebration post. System celebration posts carry ``author_person_id = None`` — nobody "wrote" them — so every author-based moderation branch evaluated to False for the person the post is about. Net effect before this branch existed: the birthday person had no delete lever over an inappropriate comment on their own birthday post (only global_admin / HR did), while every ordinary user post gave its author exactly that lever. Being the subject of a post is the same standing as having written it. Aggregate banners carry ``celebration_id = None`` and are about many people, so they correctly match nobody here. """ cel = post.celebration if post.celebration_id is None or cel is None: return False return str(cel.person_id) == str(viewer.id) def _can_delete_post( self, post: FeedPost, viewer: IdentityPerson, ) -> bool: if _is_admin(viewer): return True if ( post.author_person_id is not None and str(post.author_person_id) == str(viewer.id) ): return True # The celebrant owns their own celebration post (see # ``_is_celebrant_of``) — they may take it down. if self._is_celebrant_of(post, viewer): return True return False def _can_delete_comment( self, comment: FeedComment, post: FeedPost, viewer: IdentityPerson, ) -> bool: if _is_admin(viewer): return True if str(comment.author_person_id) == str(viewer.id): return True # Author of the post can delete others' comments on it. if ( post.author_person_id is not None and str(post.author_person_id) == str(viewer.id) ): return True # …and so can the celebrant on their own celebration post, which has # no author at all (see ``_is_celebrant_of``). if self._is_celebrant_of(post, viewer): return True return False # ──────────────────────────────────────────────────────────── # Feed listing # ──────────────────────────────────────────────────────────── def _scope_conditions( self, *, viewer: IdentityPerson, tab: str, category: str | None, ) -> tuple[list[Any], bool]: """Tab + category predicate shared by ``list_posts`` and ``search_posts``. Returns ``(conditions, unsatisfiable)``. ``unsatisfiable=True`` means the scope can never match anything (the 部门动态 tab for a person with no primary department) — the caller returns an empty page rather than issuing a query that would match everything. Extracted so search cannot drift away from listing: the search box lives INSIDE the tab bar, so a search run from 工作计划 that returned every public post in the company was answering a question the user did not ask (the tab highlight never moved). """ conds: list[Any] = [FeedPost.deleted_at.is_(None)] viewer_dept = _resolve_dept(viewer) # System posts (birthday / anniversary / aggregate banners) are # the company-wide "everyone's celebrations" stream — they belong # ONLY on the unscoped public tab. Earlier versions leaked them # into every tab on the theory "users always want to see who's # celebrating", but the cost was that 工作计划 / 部门动态 / 其它 # tabs got polluted with content that didn't match their label. # 2026-06-02 user feedback: tabs must mean what they say. include_system = (tab == USER_TAB_PUBLIC and not category) if not include_system: conds.append(FeedPost.post_type == POST_TYPE_USER) if tab == "anonymous": # A dedicated company-wide tree hole. Department scope would make # the author statistically identifiable, so anonymous posts are # always public and never inherit a department filter. conds.extend([ FeedPost.post_type == POST_TYPE_USER, FeedPost.author_visibility == ANONYMITY_ANONYMOUS, ]) elif tab == USER_TAB_MINE: conds.append(FeedPost.author_person_id == str(viewer.id)) elif tab == USER_TAB_DEPARTMENT: if viewer_dept is None: # No department membership → empty result (not an error) return conds, True # Strict dept scope — the author must be in the viewer's # department, whatever the post's visibility (#2146). # # The old first branch was a bare # ``visibility == VISIBILITY_PUBLIC`` with NO author-dept # condition, and ``department_snapshot`` was only written on # dept-visibility posts, so this tab returned every public # post in the company: an Eng employee saw Sales, HR and # Finance, and each post appeared in BOTH 公开动态 and # 部门动态. The comment above it said "strict dept scope" — # comment and code disagreed, and the code was the wrong one # (2026-06-02 feedback: tabs must mean what they say). # # Posts created before the snapshot fix carry NULL, so they # fall back to the author's CURRENT primary department — # otherwise the tab would empty itself on deploy. legacy_author_dept = ( select(IdentityOrgMembership.person_id) .where( IdentityOrgMembership.org_node_name == viewer_dept, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) ) conds.append( or_( FeedPost.department_snapshot == viewer_dept, and_( FeedPost.department_snapshot.is_(None), FeedPost.author_person_id.in_(legacy_author_dept), ), ) ) else: # PUBLIC conds.append( or_( # ``include_system`` keeps system-post rows in the # candidate set; their visibility column is unused # for them. User posts gate as usual. FeedPost.post_type.in_(list(SYSTEM_POST_TYPES)), FeedPost.visibility == VISIBILITY_PUBLIC, and_( FeedPost.visibility == VISIBILITY_DEPARTMENT, FeedPost.department_snapshot == viewer_dept if viewer_dept is not None else false(), ), ) ) if category: # Category tabs (工作计划 / 其它) are content-category views # for user posts — strict match, no system-post leakage. conds.append(FeedPost.category == category) return conds, False async def list_posts( self, *, viewer: IdentityPerson, tab: str = USER_TAB_PUBLIC, category: str | None = None, cursor: str | None = None, limit: int = 20, ) -> tuple[list[FeedPost], str | None]: """Return ``(posts, next_cursor)``. Cursor format: ``"|"``. """ if tab not in USER_TABS: tab = USER_TAB_PUBLIC conds, unsatisfiable = self._scope_conditions( viewer=viewer, tab=tab, category=category, ) if unsatisfiable: return [], None q = select(FeedPost).where(*conds) if cursor: try: cursor_dt_s, cursor_id = cursor.split("|", 1) cursor_dt = datetime.fromisoformat(cursor_dt_s) q = q.where( or_( FeedPost.created_at < cursor_dt, and_( FeedPost.created_at == cursor_dt, FeedPost.id < cursor_id, ), ) ) except (ValueError, AttributeError): pass q = q.order_by( desc(FeedPost.is_pinned), desc(FeedPost.created_at), desc(FeedPost.id), ).limit(limit + 1) rows = (await self.db.execute(q)).scalars().all() next_cursor: str | None = None if len(rows) > limit: tail = rows[limit - 1] next_cursor = f"{tail.created_at.isoformat()}|{tail.id}" rows = rows[:limit] return list(rows), next_cursor async def hydrate_posts( self, posts: list[FeedPost], *, viewer: IdentityPerson, ) -> list[dict[str, Any]]: """Return list of FeedPostDTO-shaped dicts.""" if not posts: return [] post_ids = [p.id for p in posts] author_ids = { str(p.author_person_id) for p in posts if p.author_person_id is not None } # Mentions mention_rows = (await self.db.execute( select(FeedMention).where(FeedMention.post_id.in_(post_ids)) )).scalars().all() mention_ids = {str(m.mentioned_person_id) for m in mention_rows} # Celebrations (for system posts) — already eager-loaded via # relationship in some paths, but ensure for aggregate types. aggregate_dates: set[date] = set() banner_date_by_post: dict[str, date] = {} for p in posts: if p.post_type in { POST_TYPE_AGGREGATE_BIRTHDAY, POST_TYPE_AGGREGATE_ANNIVERSARY, }: # Aggregate posts don't link to a single celebration, # but we attach the day's celebrants by occasion_date. banner_date = _banner_occasion_date(p) if banner_date is None: continue banner_date_by_post[p.id] = banner_date if p.celebration is None: aggregate_dates.add(banner_date) # Build celebrant lookup per (date, type). celebrant_index: dict[ tuple[date, str], list[FeedCelebration] ] = {} if aggregate_dates: cels = (await self.db.execute( select(FeedCelebration).where( FeedCelebration.occasion_date.in_(aggregate_dates), ) )).scalars().all() for c in cels: celebrant_index.setdefault( (c.occasion_date, c.celebration_type), [], ).append(c) # Person-celebration backref so aggregate banners can render. for p in posts: if p.celebration is not None: author_ids.add(str(p.celebration.person_id)) # Likes for viewer liked_post_ids: set[str] = set() if post_ids: like_rows = (await self.db.execute( select(FeedLike.post_id).where( FeedLike.post_id.in_(post_ids), FeedLike.person_id == str(viewer.id), ) )).all() liked_post_ids = {row[0] for row in like_rows} persons = await self._hydrate_persons(author_ids | mention_ids) # Attachments — pre-sign. Storage config (bucket/region/prefix) # is identical for every attachment in this batch, so resolve # ONCE and reuse the same S3 client. The previous code called # ``presign_get`` per attachment, which internally re-fetched # 3 PlatformSetting rows per call — a 9-image post fired 27 # extra DB hits during hydration. Now config is resolved once; # ``generate_presigned_url`` itself is CPU-only (HMAC signing). from app.core.storage import ( # local import — avoid top-level StorageNotConfiguredError, _get_s3_client, resolve_storage_config, ) attachments_by_post: dict[str, list[dict[str, Any]]] = {} att_rows = (await self.db.execute( select(FeedAttachment) .where(FeedAttachment.post_id.in_(post_ids)) .order_by(FeedAttachment.post_id, FeedAttachment.position) )).scalars().all() s3_client = None s3_bucket = None if att_rows: try: cfg = await resolve_storage_config(self.db) s3_client = _get_s3_client(cfg["region"]) s3_bucket = cfg["bucket"] except StorageNotConfiguredError: logger.exception("feed.attachment.storage_config_missing") from app.storage_router import FEED_PURPOSES, parse_object_key for a in att_rows: url: str | None = None # Re-check before signing. The create path now refuses a # foreign key, but rows written before that gate existed are # still in the table, and this is the line that actually # mints the credential — an attachment we would not accept # today must not be signed today either (#2124). parsed_key = parse_object_key(a.s3_key or "") if parsed_key is None or parsed_key[0] not in FEED_PURPOSES: logger.warning( "feed.attachment.refused_presign post=%s key=%s", a.post_id, a.s3_key, ) attachments_by_post.setdefault(a.post_id, []).append({ "s3_key": a.s3_key, "mime_type": a.mime_type, "width": a.width, "height": a.height, "download_url": None, }) continue if s3_client is not None and s3_bucket: try: url = s3_client.generate_presigned_url( "get_object", Params={"Bucket": s3_bucket, "Key": a.s3_key}, ExpiresIn=600, ) except Exception: logger.exception( "feed.attachment.presign_failed post=%s key=%s", a.post_id, a.s3_key, ) attachments_by_post.setdefault(a.post_id, []).append({ "s3_key": a.s3_key, "mime_type": a.mime_type, "width": a.width, "height": a.height, "download_url": url, }) # Group mentions per post mentions_by_post: dict[str, list[str]] = {} for m in mention_rows: if m.post_id: mentions_by_post.setdefault(m.post_id, []).append( str(m.mentioned_person_id), ) out: list[dict[str, Any]] = [] for p in posts: author = ( persons.get(str(p.author_person_id)) if p.author_person_id else None ) mentioned = [ self._person_lite(persons.get(pid)) for pid in mentions_by_post.get(p.id, []) if persons.get(pid) ] celebration_lite: dict[str, Any] | None = None if p.celebration is not None: target = persons.get(str(p.celebration.person_id)) celebration_lite = { "person_id": str(p.celebration.person_id), "celebration_id": p.celebration.id, "display_name": ( p.celebration.display_name_snapshot or (target.display_name if target else None) ), "department": ( p.celebration.department_snapshot or (target.primary_department if target else None) ), "avatar_url": ( p.celebration.avatar_url_snapshot or (target.avatar_url if target else None) ), "years": p.celebration.years, "illustration_key": p.celebration.illustration_key, } aggregate_celebrants: list[dict[str, Any]] = [] if p.post_type in { POST_TYPE_AGGREGATE_BIRTHDAY, POST_TYPE_AGGREGATE_ANNIVERSARY, }: ctype = ( "birthday" if p.post_type == POST_TYPE_AGGREGATE_BIRTHDAY else "anniversary" ) banner_date = banner_date_by_post.get(p.id) day_celebrants = ( celebrant_index.get((banner_date, ctype), []) if banner_date is not None else [] ) for c in day_celebrants: aggregate_celebrants.append({ "person_id": str(c.person_id), "celebration_id": c.id, "display_name": c.display_name_snapshot, "department": c.department_snapshot, "avatar_url": c.avatar_url_snapshot, "years": c.years, "illustration_key": c.illustration_key, }) out.append({ "id": p.id, "post_type": p.post_type, "visibility": p.visibility, "category": p.category, "author_visibility": p.author_visibility, "body": p.body, "is_pinned": p.is_pinned, # Do not merely hide this in the UI: an anonymous feed DTO # must never carry PII to ordinary clients (or browser dev # tools, logs and extension code). Moderation has a separate, # audited reveal endpoint. "author": ( None if p.author_visibility == ANONYMITY_ANONYMOUS else self._person_lite(author) ), "created_at": _iso_utc(p.created_at), "last_edited_at": _iso_utc(p.last_edited_at), "edit_count": p.edit_count or 0, "like_count": p.like_count, "comment_count": p.comment_count, "liked_by_me": p.id in liked_post_ids, "can_delete": self._can_delete_post(p, viewer), "can_edit": self._can_edit_post(p, viewer), "attachments": attachments_by_post.get(p.id, []), "mentioned_persons": [m for m in mentioned if m is not None], "celebration": celebration_lite, "aggregate_celebrants": aggregate_celebrants, }) return out # ──────────────────────────────────────────────────────────── # Post create / delete # ──────────────────────────────────────────────────────────── async def create_user_post( self, *, author: IdentityPerson, body: str, visibility: str, category: str, mentioned_person_ids: list[str], attachments: list[dict[str, Any]], client_token: str | None = None, author_visibility: str = ANONYMITY_NAMED, ) -> tuple[FeedPost, bool]: """Publish a user post. Returns ``(post, created)``. ``created`` is False when ``client_token`` replayed an earlier call, in which case the stored post is returned untouched and the caller must NOT re-audit or re-fan-out mentions — that is the whole point. The lookup is a LOCKING read, for the reason documented at length on ``_find_thanks_post``: MySQL's REPEATABLE READ answers a plain SELECT from the snapshot taken at the request's first query, so two racing requests would both see "no prior post" no matter how carefully the check is written. The unique index is the backstop — if a third racer slips between the locking read and the flush, the IntegrityError is caught below and resolved into the same replay. """ token = (client_token or "").strip() or None if token is not None: existing = (await self.db.execute( select(FeedPost) .where( FeedPost.author_person_id == str(author.id), FeedPost.client_token == token, ) .limit(1) .with_for_update() )).scalar_one_or_none() if existing is not None and existing.deleted_at is None: return existing, False if existing is not None: # A TOMBSTONE holds the token. Release it and fall through to a # normal insert, rather than replaying a post the reader # deleted — the old behaviour answered 200 `replayed: true` and # hydrated the deleted row, so the year-in-review told them # "you already shared — it's below" while pointing at nothing, # and that year could never be shared again. # # Deliberately UNLIKE the thanks-post guard, which counts # soft-deleted rows on purpose so a delete-repost loop cannot # re-fan-out @-mentions to every well-wisher. Here there is no # such risk: the recap mentions nobody, and the composer mints # a fresh token per draft anyway. The row lock taken above # serialises this release against a concurrent racer. existing.client_token = None await self.db.flush() if author_visibility not in {ANONYMITY_NAMED, ANONYMITY_ANONYMOUS}: author_visibility = ANONYMITY_NAMED if author_visibility == ANONYMITY_ANONYMOUS: # Anonymous content is deliberately text-only in v1. Existing # uploads can retain EXIF, filenames or visual identifiers; do # not claim anonymity until a metadata-scrubbing pipeline exists. if attachments: raise ValidationError("anonymous posts cannot include attachments") visibility = VISIBILITY_PUBLIC mentioned_person_ids = [] if visibility not in {VISIBILITY_PUBLIC, VISIBILITY_DEPARTMENT}: visibility = VISIBILITY_PUBLIC if category not in {"general", "work_plan", "other"}: category = "general" if len(attachments) > MAX_ATTACHMENTS_PER_POST: attachments = attachments[:MAX_ATTACHMENTS_PER_POST] _validate_attachment_keys(attachments, owner_id=author.id) if len(mentioned_person_ids) > MAX_MENTIONS_PER_POST: mentioned_person_ids = mentioned_person_ids[:MAX_MENTIONS_PER_POST] author_dept = _resolve_dept(author) # A user with no primary department picking "department" scope # would post a row whose ``department_snapshot=None`` — no # other viewer would ever match (NULL doesn't equal anything), # so it'd only be visible to the author themselves. Surfacing # this as "I posted but no teammate can see it" is the wrong # default; silently widen to ``public`` instead. Author is # rare (contractors / system accounts / unfinished org-sync). if visibility == VISIBILITY_DEPARTMENT and not author_dept: visibility = VISIBILITY_PUBLIC now = datetime.now(UTC) post = FeedPost( author_person_id=str(author.id), post_type=POST_TYPE_USER, visibility=visibility, category=category, author_visibility=author_visibility, body=body, # Stamped for EVERY user post, not only dept-visibility ones # (#2146). The column means "the author's department when # they posted" — that is what the 部门动态 tab needs in order # to mean anything, and it was previously NULL on public # posts, which made the tab's dept predicate match nothing # and fall through to "every public post in the company". department_snapshot=author_dept, # Python-side timestamp so we never need a follow-up refresh. # The column's server_default=NOW() stays as a safety net for # raw-SQL inserts; ORM inserts get a definitive Python value # so the response can serialise it without a SELECT round-trip. created_at=now, updated_at=now, client_token=token, ) # SAVEPOINT: a duplicate token raises IntegrityError, and without the # nested transaction that error would poison the caller's whole # transaction — the audit write and everything after it would fail on # an already-aborted connection rather than resolving into a clean # replay. # # ``add`` goes INSIDE the savepoint. Added outside, the instance is # already pending on the outer session, so the rollback does not # detach it and the next autoflush replays the same failing INSERT # from a context that cannot await. try: async with self.db.begin_nested(): self.db.add(post) await self.db.flush() except IntegrityError: if token is None: raise existing = (await self.db.execute( select(FeedPost).where( FeedPost.author_person_id == str(author.id), FeedPost.client_token == token, ).limit(1) # LOCKING, for the same snapshot reason as the pre-check. This # branch looks unreachable on MySQL — the pre-check's gap lock # serialises the racers — but PRODUCTION IS TiDB, which has no # gap locks: both racers insert, the loser gets 1062, and a # plain SELECT here would read its pre-race snapshot, find # nothing, and re-raise into a 500 on exactly the double-click # this feature exists to absorb. .with_for_update() )).scalar_one_or_none() if existing is None or existing.deleted_at is not None: raise return existing, False from app.core.storage_uploads import mark_uploads_referenced from app.storage_router import FEED_PURPOSES claimed_uploads = await mark_uploads_referenced( self.db, object_keys=[att["s3_key"] for att in attachments], person_id=author.id, allowed_purposes=FEED_PURPOSES, resource_type="feed_post", resource_id=str(post.id), ) for pos, att in enumerate(attachments): grant = claimed_uploads[att["s3_key"]] self.db.add(FeedAttachment( post_id=post.id, s3_key=att["s3_key"], mime_type=grant.content_type, # Store the signed-and-HEAD-verified size, never the client # field carried in the create-post payload. size_bytes=grant.declared_size_bytes, width=att.get("width"), height=att.get("height"), position=pos, )) # Mentions — dedup, skip self-mention. seen: set[str] = set() for pid in mentioned_person_ids: pid = str(pid) if pid == str(author.id) or pid in seen: continue seen.add(pid) self.db.add(FeedMention( mentioned_person_id=pid, post_id=post.id, author_person_id=str(author.id), )) await self.db.flush() return post, True async def soft_delete_post( self, *, post_id: str, viewer: IdentityPerson, ) -> None: post = await self.db.get(FeedPost, post_id) if post is None or post.deleted_at is not None: raise NotFoundError("post not found") if not self._can_delete_post(post, viewer): raise ForbiddenError("cannot delete this post") post.deleted_at = datetime.now(UTC) await self.db.flush() # ──────────────────────────────────────────────────────────── # Comments # ──────────────────────────────────────────────────────────── async def add_comment( self, *, post_id: str, author: IdentityPerson, body: str, mentioned_person_ids: list[str], reply_to_comment_id: str | None, author_visibility: str = ANONYMITY_NAMED, sticker_key: str | None = None, blessing_id: str | None = None, ) -> FeedComment: post = await self.db.get(FeedPost, post_id) if post is None or post.deleted_at is not None: raise NotFoundError("post not found") if not self._can_view_post(post, author): raise ForbiddenError("cannot comment on this post") if post.author_visibility == ANONYMITY_ANONYMOUS: author_visibility = ANONYMITY_ANONYMOUS mentioned_person_ids = [] elif author_visibility not in {ANONYMITY_NAMED, ANONYMITY_ANONYMOUS}: author_visibility = ANONYMITY_NAMED # Validate the reply pointer belongs to THIS post and still exists — # an unchecked id stored verbatim produces dangling / cross-thread # reply references. Drop the pointer (render as a top-level comment) # rather than persist an invalid one. The three rejection cases are # the complete set for a NEW comment: unknown id, soft-deleted # parent, parent on another post. (Self-reference is unreachable — # the id is minted below, after this check, so no caller can name # it.) The effective value is echoed back in the create response so # the client can tell a dropped pointer from an accepted one instead # of rendering a reply that the server stored as top-level. if reply_to_comment_id is not None: parent = await self.db.get(FeedComment, reply_to_comment_id) if ( parent is None or getattr(parent, "deleted_at", None) is not None or str(parent.post_id) != str(post.id) ): reply_to_comment_id = None now = datetime.now(UTC) comment = FeedComment( post_id=post.id, author_person_id=str(author.id), body=body, author_visibility=author_visibility, reply_to_comment_id=reply_to_comment_id, sticker_key=sticker_key, blessing_id=blessing_id, # Python-side timestamp — see ``create_user_post`` rationale. # Without this, ``comment.created_at`` stays None until a # refresh and the router endpoint's ``isoformat()`` raises # AttributeError → 500 AFTER the row was already committed # (the bug surfaced as "评论按钮报错但已写入数据库"). created_at=now, updated_at=now, ) self.db.add(comment) await self.db.flush() # Atomic increment — read-modify-write on ``post.comment_count`` # races against concurrent commenters and silently loses # updates (both flushes commit N+1 instead of N+2). The UPDATE # below evaluates the +1 server-side so concurrent writes are # serialised by the row lock. ``synchronize_session=False`` skips # ORM identity-map reconcile (cheap and we don't read post. # comment_count again in this scope). await self.db.execute( update(FeedPost) .where(FeedPost.id == post.id) .values(comment_count=FeedPost.comment_count + 1), ) # Cap stored mentions per comment (mirrors create_user_post) — bounds # both the FeedMention row count and the Feishu mention-DM fan-out so a # single comment can't write/notify an unbounded list. if len(mentioned_person_ids) > MAX_MENTIONS_PER_POST: mentioned_person_ids = mentioned_person_ids[:MAX_MENTIONS_PER_POST] seen: set[str] = set() for pid in mentioned_person_ids: pid = str(pid) if pid == str(author.id) or pid in seen: continue seen.add(pid) self.db.add(FeedMention( mentioned_person_id=pid, comment_id=comment.id, author_person_id=str(author.id), )) await self.db.flush() return comment async def list_comments( self, *, post_id: str, viewer: IdentityPerson, limit: int = 200, ) -> list[dict[str, Any]]: post = await self.db.get(FeedPost, post_id) if post is None or post.deleted_at is not None: raise NotFoundError("post not found") if not self._can_view_post(post, viewer): raise ForbiddenError("cannot view this post") # Bound the result so a hot post with thousands of comments can't # force an unbounded materialization (the /posts list is already # le=50-capped). Fetch the newest ``limit`` rows, then present them # oldest-first so the chronological reading order is preserved for # the common case (posts with ≤limit comments are unaffected). rows = list((await self.db.execute( select(FeedComment) .where( FeedComment.post_id == post.id, FeedComment.deleted_at.is_(None), ) .order_by(FeedComment.created_at.desc()) .limit(max(1, limit)) )).scalars().all()) rows.reverse() author_ids = {str(c.author_person_id) for c in rows} persons = await self._hydrate_persons(author_ids) return [ { "id": c.id, "body": c.body, "author_visibility": c.author_visibility, "author": ( None if c.author_visibility == ANONYMITY_ANONYMOUS else self._person_lite(persons.get(str(c.author_person_id))) ), "created_at": _iso_utc(c.created_at), "last_edited_at": _iso_utc(c.last_edited_at), "edit_count": c.edit_count or 0, "reply_to_comment_id": c.reply_to_comment_id, "sticker_key": c.sticker_key, "can_delete": self._can_delete_comment(c, post, viewer), "can_edit": self._can_edit_comment(c, viewer), } for c in rows ] async def soft_delete_comment( self, *, comment_id: str, viewer: IdentityPerson, ) -> tuple[FeedComment, str]: """Soft-delete a comment. Returns ``(comment, prev_body)`` so the router can audit-log the body that was just hidden — admins need the pre-delete content for forensic review, and the row's ``body`` column is intentionally preserved (only ``deleted_at`` flips). """ comment = await self.db.get(FeedComment, comment_id) if comment is None or comment.deleted_at is not None: raise NotFoundError("comment not found") post = await self.db.get(FeedPost, comment.post_id) if post is None: raise NotFoundError("post not found") if not self._can_delete_comment(comment, post, viewer): raise ForbiddenError("cannot delete this comment") prev_body = comment.body comment.deleted_at = datetime.now(UTC) # Atomic decrement (see ``add_comment``). ``GREATEST`` guards # against ever going negative if a race deletes the same # comment row twice (shouldn't happen — soft-delete is # idempotent at the row level — but cheap insurance). await self.db.execute( update(FeedPost) .where(FeedPost.id == post.id) .values( comment_count=func.greatest( FeedPost.comment_count - 1, 0, ), ), ) # Cascade soft-delete the matching blessing row so the # celebration page's 留言板 stays consistent with the comment # thread. ``mirror_comment_id`` was set by ``send_blessing``. if comment.blessing_id: blessing = await self.db.get(FeedBlessing, comment.blessing_id) if blessing is not None: # Hard-delete the blessing row — soft-delete would need # a ``deleted_at`` column we didn't add. The UNIQUE on # (celebration_id, sender_person_id) is intentionally # released so the same sender can re-bless if they # change their mind after deleting. await self.db.delete(blessing) await self.db.flush() return comment, prev_body # ──────────────────────────────────────────────────────────── # Community reports / controlled identity disclosure # ──────────────────────────────────────────────────────────── async def create_report( self, *, resource_type: str, resource_id: str, reporter: IdentityPerson, reason: str, detail: str | None, ) -> FeedReport: if resource_type == "post": target = await self.db.get(FeedPost, resource_id) if target is None or target.deleted_at is not None: raise NotFoundError("post not found") if not self._can_view_post(target, reporter): raise ForbiddenError("cannot report this post") elif resource_type == "comment": comment_target = ( await self.db.execute( select(FeedComment).where(FeedComment.id == resource_id) ) ).scalar_one_or_none() if comment_target is None or comment_target.deleted_at is not None: raise NotFoundError("comment not found") post = await self.db.get(FeedPost, comment_target.post_id) if post is None or not self._can_view_post(post, reporter): raise ForbiddenError("cannot report this comment") else: raise ValidationError("unsupported feed report resource") report = FeedReport( resource_type=resource_type, resource_id=resource_id, reporter_person_id=str(reporter.id), reason=reason, detail=(detail or "").strip() or None, ) try: async with self.db.begin_nested(): self.db.add(report) await self.db.flush() except IntegrityError as exc: raise ConflictError("you already reported this content") from exc return cast(FeedReport, report) async def list_reports(self, *, status: str | None = None) -> list[FeedReport]: stmt = select(FeedReport).order_by(FeedReport.created_at.desc()).limit(500) if status in {"open", "resolved", "dismissed"}: stmt = stmt.where(FeedReport.status == status) return list((await self.db.execute(stmt)).scalars().all()) async def resolve_report( self, *, report_id: str, moderator: IdentityPerson, status: str, hide_content: bool, note: str | None, ) -> FeedReport: report = await self.db.get(FeedReport, report_id, with_for_update=True) if report is None: raise NotFoundError("feed report not found") if hide_content: if report.resource_type == "post": post = await self.db.get(FeedPost, report.resource_id) if post is not None and post.deleted_at is None: post.deleted_at = datetime.now(UTC) else: comment = await self.db.get(FeedComment, report.resource_id) if comment is not None and comment.deleted_at is None: comment.deleted_at = datetime.now(UTC) await self.db.execute( update(FeedPost).where(FeedPost.id == comment.post_id).values( comment_count=func.greatest(FeedPost.comment_count - 1, 0), ), ) report.status = status report.resolved_by_person_id = str(moderator.id) report.resolution_note = (note or "").strip() or None report.resolved_at = datetime.now(UTC) await self.db.flush() return cast(FeedReport, report) async def reveal_report_author( self, *, report_id: str, ) -> IdentityPerson: report = await self.db.get(FeedReport, report_id) if report is None: raise NotFoundError("feed report not found") author_id: str | None = None if report.resource_type == "post": post = await self.db.get(FeedPost, report.resource_id) author_id = str(post.author_person_id) if post else None elif report.resource_type == "comment": comment = await self.db.get(FeedComment, report.resource_id) author_id = str(comment.author_person_id) if comment else None if author_id is None: raise NotFoundError("reported content author not found") author = await self.db.get(IdentityPerson, author_id) if author is None: raise NotFoundError("reported content author not found") return author # ──────────────────────────────────────────────────────────── # Edit (post / comment) # ──────────────────────────────────────────────────────────── def _can_edit_post( self, post: FeedPost, viewer: IdentityPerson, ) -> bool: """Only the author can edit their own user post. Admins do NOT get an edit lever — admins can soft-delete via the existing moderation path, but rewriting another user's words would be worse for trust than removing the row. System posts (birthday / anniversary / aggregate banners) are never editable. """ if post.post_type != POST_TYPE_USER: return False if post.deleted_at is not None: return False return ( post.author_person_id is not None and str(post.author_person_id) == str(viewer.id) ) def _can_edit_comment( self, comment: FeedComment, viewer: IdentityPerson, ) -> bool: if comment.deleted_at is not None: return False # Mirror comments (sent via a blessing) are tied to a structured # event — editing the text out from under that contract makes the # 留言板 inconsistent. Author-edits only on plain comments. if comment.blessing_id is not None: return False return str(comment.author_person_id) == str(viewer.id) async def edit_post( self, *, post_id: str, viewer: IdentityPerson, new_body: str, ) -> tuple[FeedPost, str]: """Rewrite ``post.body``. Returns ``(post, prev_body)`` for the caller to log into the audit trail. Raises if the viewer is not the author, the post is system-typed, or the post is deleted. """ post = await self.db.get(FeedPost, post_id) if post is None or post.deleted_at is not None: raise NotFoundError("post not found") if not self._can_edit_post(post, viewer): raise ForbiddenError("cannot edit this post") new_body = new_body.strip() if not new_body: raise ForbiddenError("body cannot be empty") prev_body = post.body if prev_body == new_body: # No-op edit — don't bump counters / timestamps. return post, prev_body now = datetime.now(UTC) post.body = new_body post.last_edited_at = now # Mirror server-side ``onupdate=NOW()`` python-side so the # response serializer reads a fresh in-memory value (consistent # with create_user_post / add_comment / create_announcement). post.updated_at = now post.edit_count = (post.edit_count or 0) + 1 await self.db.flush() return post, prev_body async def edit_comment( self, *, comment_id: str, viewer: IdentityPerson, new_body: str, ) -> tuple[FeedComment, str]: comment = await self.db.get(FeedComment, comment_id) if comment is None or comment.deleted_at is not None: raise NotFoundError("comment not found") if not self._can_edit_comment(comment, viewer): raise ForbiddenError("cannot edit this comment") new_body = new_body.strip() if not new_body: raise ForbiddenError("body cannot be empty") prev_body = comment.body if prev_body == new_body: return comment, prev_body now = datetime.now(UTC) comment.body = new_body comment.last_edited_at = now comment.updated_at = now comment.edit_count = (comment.edit_count or 0) + 1 await self.db.flush() return comment, prev_body # ──────────────────────────────────────────────────────────── # Likes # ──────────────────────────────────────────────────────────── async def toggle_like( self, *, post_id: str, person: IdentityPerson, like: bool, ) -> tuple[int, bool]: """Return ``(like_count, changed)``. ``changed`` is True only when a like row was actually inserted (like=True) or deleted (like=False) — callers gate the "X liked your post" notification on it so repeated like calls (double-click, LLM retry loop) don't spam the author. """ post = await self.db.get(FeedPost, post_id) if post is None or post.deleted_at is not None: raise NotFoundError("post not found") if not self._can_view_post(post, person): raise ForbiddenError("cannot interact with this post") existing = (await self.db.execute( select(FeedLike).where( FeedLike.post_id == post.id, FeedLike.person_id == str(person.id), ) )).scalar_one_or_none() changed = False if like: if existing is None: async with self.db.begin_nested(): try: self.db.add(FeedLike( post_id=post.id, person_id=str(person.id), )) await self.db.flush() except IntegrityError: # Concurrent insert — already liked. Bail out # without bumping the count (the other tx did). # SAVEPOINT rollback keeps the outer session # usable so the caller's commit still runs. return post.like_count or 0, False # Atomic increment — see ``add_comment``. await self.db.execute( update(FeedPost) .where(FeedPost.id == post.id) .values(like_count=FeedPost.like_count + 1), ) changed = True else: if existing is not None: # Race-safe unlike: delete via a statement whose rowcount tells # us whether THIS transaction actually removed the row. Two # concurrent unlikes (double-tap / retry) both read the same # FeedLike under MVCC; the bare ``db.delete`` + unconditional # decrement let each drop like_count by 1 → count off by 2 for # a single like removed. Only decrement when we truly deleted. result = await self.db.execute( delete(FeedLike).where( FeedLike.post_id == post.id, FeedLike.person_id == str(person.id), ) ) if getattr(result, "rowcount", 0): await self.db.execute( update(FeedPost) .where(FeedPost.id == post.id) .values( like_count=func.greatest( FeedPost.like_count - 1, 0, ), ), ) changed = True await self.db.flush() # Re-read so the response carries the post-commit value (the # in-memory ``post.like_count`` no longer reflects truth after # the bare-SQL UPDATE). await self.db.refresh(post, attribute_names=["like_count"]) return post.like_count or 0, changed # ──────────────────────────────────────────────────────────── # Search # ──────────────────────────────────────────────────────────── async def search_posts( self, *, viewer: IdentityPerson, query: str, tab: str = USER_TAB_PUBLIC, category: str | None = None, limit: int = 30, ) -> list[FeedPost]: """LIKE search, scoped to the SAME tab / category the caller is looking at. The search box is rendered inside the tab bar, so a search issued from 工作计划 that returned every public post in the company was answering a different question than the one the UI asked — and the tab highlight never moved to say so. ``tab`` / ``category`` default to the unscoped public view, which is the old behaviour, so callers that don't pass them (MCP ``feed_search_posts``) are unaffected. The old query also carried a standalone ``author_person_id == viewer`` clause. It is dropped rather than re-added: every own post is already reachable — public posts via the public predicate, department posts via the snapshot (which for your own post is by definition your own department) — and ``tab='mine'`` is the precise way to ask for only your own. """ if tab not in USER_TABS: tab = USER_TAB_PUBLIC conds, unsatisfiable = self._scope_conditions( viewer=viewer, tab=tab, category=category, ) if unsatisfiable: return [] # LIKE with ESCAPE for safety. pattern = "%" + query.replace("\\", "\\\\")\ .replace("%", "\\%").replace("_", "\\_") + "%" q = ( select(FeedPost) .where(*conds, FeedPost.body.like(pattern, escape="\\")) .order_by(desc(FeedPost.created_at)) .limit(limit) ) return list((await self.db.execute(q)).scalars().all()) # ──────────────────────────────────────────────────────────── # Celebrations & blessings # ──────────────────────────────────────────────────────────── async def get_today_celebrations( self, *, today: date, ) -> tuple[list[FeedCelebration], list[FeedCelebration]]: rows = (await self.db.execute( select(FeedCelebration) .where(FeedCelebration.occasion_date == today) .order_by(FeedCelebration.celebration_type, FeedCelebration.years) )).scalars().all() birthdays = [c for c in rows if c.celebration_type == "birthday"] anniversaries = [ c for c in rows if c.celebration_type == "anniversary" ] return birthdays, anniversaries async def get_celebration( self, celebration_id: str, ) -> FeedCelebration: c = await self.db.get(FeedCelebration, celebration_id) if c is None: raise NotFoundError("celebration not found") return c async def _celebrant_today(self, celebration: FeedCelebration) -> date: """"Today" as the CELEBRANT experiences it. ``occasion_date`` is seeded by the cron in the celebrant's own timezone (``resolve_timezone(person)``), so any window arithmetic against it has to use the same clock — comparing a Shanghai-seeded date against a US sender's local date shifts the whole window by a day at the boundary. """ from app.core.working_hours import resolve_timezone persons = await self._hydrate_persons({str(celebration.person_id)}) person = persons.get(str(celebration.person_id)) tz = resolve_timezone(person) if person is not None else UTC return datetime.now(tz).date() async def _assert_blessing_window_open( self, celebration: FeedCelebration, ) -> None: """Blessings are only accepted for ``[today - 7d, today]``. Celebration ids are enumerable by any authenticated user and ``send_blessing`` fires a Feishu push at the celebrant, so without this any employee could open a birthday from three years ago and make the celebrant's phone buzz with a fresh "happy birthday" — the page renders identically whatever the date. The upper bound matters too: a celebration seeded ahead of its day must not collect blessings before the day arrives. """ today = await self._celebrant_today(celebration) earliest = today - timedelta(days=BLESSING_WINDOW_DAYS) if not (earliest <= celebration.occasion_date <= today): raise ConflictError( "this celebration is no longer accepting blessings", code=ERR_BLESSING_WINDOW_CLOSED, ) async def send_blessing( self, *, celebration_id: str, sender: IdentityPerson, sticker_key: str, message: str, ) -> FeedBlessing: celebration = await self.get_celebration(celebration_id) if str(celebration.person_id) == str(sender.id): raise ForbiddenError("you can't send a blessing to yourself") await self._assert_blessing_window_open(celebration) # Pre-check existence so the duplicate case returns 409 without # spending an INSERT + rollback on the wire. The UNIQUE in the # SAVEPOINT below still defends against a concurrent duplicate # (TOCTOU between this read and our flush). prior = (await self.db.execute( select(FeedBlessing).where( FeedBlessing.celebration_id == celebration.id, FeedBlessing.sender_person_id == str(sender.id), ) )).scalar_one_or_none() if prior is not None: raise ConflictError("you've already sent a blessing") # Locate the system post for this celebration so the mirrored # comment lands on the right thread. system_post = (await self.db.execute( select(FeedPost).where( FeedPost.celebration_id == celebration.id, FeedPost.post_type.in_([ POST_TYPE_SYSTEM_BIRTHDAY, POST_TYPE_SYSTEM_ANNIVERSARY, ]), FeedPost.deleted_at.is_(None), ) )).scalar_one_or_none() if system_post is None: raise NotFoundError("celebration system post not found") # Race window: another tab / device sent a blessing between # our pre-check and this flush. The WHOLE flow (mirror + mention # + blessing) must roll back together — if only the blessing # row is reverted, the mirror comment stays on the system post # without a backing blessing row → ghost comment on the 留言板. # SAVEPOINT around the entire write set, so a UNIQUE collision # on (celebration_id, sender_person_id) rolls back all three. # # Python-side ``created_at`` / ``updated_at`` on both rows — # same fix as ``add_comment`` / ``create_user_post`` (PR #658): # without it, the columns rely on ``server_default=NOW()``, the # values land in the DB but the in-session instance keeps them # un-loaded; the router's response serialisation # (``_iso_utc(blessing.created_at)``) then triggers an attribute # refresh after ``db.commit()`` which raises MissingGreenlet in # async context → 500 AFTER the blessing was already committed # (surfaces as "send button errors, but the blessing already # arrived" — the exact UX in this round). now = datetime.now(UTC) try: async with self.db.begin_nested(): mirror = FeedComment( post_id=system_post.id, author_person_id=str(sender.id), body=message, sticker_key=sticker_key, created_at=now, updated_at=now, ) self.db.add(mirror) await self.db.flush() # need mirror.id for blessing FK # Always mention the celebrant so the celebration page # renders "@xxx ..." (matches screenshot 4). self.db.add(FeedMention( mentioned_person_id=str(celebration.person_id), comment_id=mirror.id, author_person_id=str(sender.id), )) blessing = FeedBlessing( celebration_id=celebration.id, sender_person_id=str(sender.id), sticker_key=sticker_key, message=message, mirror_comment_id=mirror.id, created_at=now, ) self.db.add(blessing) await self.db.flush() except IntegrityError as exc: raise ConflictError("you've already sent a blessing") from exc # Back-link from comment to blessing for cascade soft-delete. mirror.blessing_id = blessing.id # Atomic increment of the system post's comment_count — same # race-condition reasoning as ``add_comment``. await self.db.execute( update(FeedPost) .where(FeedPost.id == system_post.id) .values(comment_count=FeedPost.comment_count + 1), ) await self.db.flush() return blessing async def list_blessings( self, *, celebration_id: str, ) -> list[dict[str, Any]]: rows = (await self.db.execute( select(FeedBlessing) .where(FeedBlessing.celebration_id == celebration_id) .order_by(desc(FeedBlessing.created_at)) )).scalars().all() sender_ids = {str(r.sender_person_id) for r in rows} persons = await self._hydrate_persons(sender_ids) return [ { "id": r.id, "celebration_id": r.celebration_id, "sender": self._blessing_sender( persons.get(str(r.sender_person_id)), ), "sticker_key": r.sticker_key, "message": r.message, "created_at": _iso_utc(r.created_at), # Drives the 已致谢 marker on each blessing card. NULL until # the celebrant acknowledges this specific blessing (either # one-by-one or via the bulk 回谢 flow). "acked_at": _iso_utc(r.acked_at), } for r in rows ] # ── Celebrant → blesser reply loop ─────────────────────────── # # Before this, a celebration page was write-only for everyone EXCEPT # the celebrant: colleagues could send blessings, the celebrant could # only read them (the CTA slot rendered a grey italic line). The two # entry points below are the reply half of that loop — per-blessing # acknowledgement, and one 回谢 post that @-mentions everyone who # wrote, reusing the ordinary mention-notification chain. async def _require_celebrant( self, celebration_id: str, viewer: IdentityPerson, *, lock: bool = False, ) -> FeedCelebration: """Load the celebration and assert ``viewer`` is who it is about. Ownership is checked BEFORE any child row is read so a non-celebrant probing blessing ids gets a flat 403 and learns nothing about which ids exist. ``lock=True`` takes a row lock on the celebration, which serialises every write that claims "once per celebration". ``thank_all_blessings`` needs it: its guard is a read-then-write (look for a prior thanks post, then insert one), and two concurrent requests both read "no prior post" and both insert — verified in review to produce two feed posts and two rounds of @-mention pushes to every well-wisher. A unique index would be the other option, but MySQL has no partial unique index and the constraint only applies to one post_type, so the lock is the honest fix. """ stmt = select(FeedCelebration).where(FeedCelebration.id == celebration_id) if lock: stmt = stmt.with_for_update() celebration = (await self.db.execute(stmt)).scalar_one_or_none() if celebration is None: raise NotFoundError("celebration not found") if str(celebration.person_id) != str(viewer.id): raise ForbiddenError("only the celebrant can reply here") return celebration async def ack_blessing( self, *, celebration_id: str, blessing_id: str, viewer: IdentityPerson, ) -> tuple[FeedBlessing, bool]: """Mark one blessing as acknowledged. Returns ``(blessing, changed)``. Idempotent by design, NOT by 409: this is a tap-to-thank affordance at the same weight as a like, and a double tap / offline retry must read as success. ``changed`` is False on the replay so the caller can skip the audit write. The blessing is loaded under a ROW LOCK for the same reason ``thank_all_blessings`` locks (see ``_require_celebrant``): the ``acked_at is None`` test below is a read-then-write, and MySQL's REPEATABLE READ answers a plain read from the snapshot taken at the request's first query — ``get_current_user``, long before this runs. Two taps arriving together therefore both saw ``acked_at is None``, both returned ``changed=True``, and both wrote an audit row for a single acknowledgement. Locking the row serialises them, so the replay reads the committed ``acked_at`` and reports ``changed=False``. """ celebration = await self._require_celebrant(celebration_id, viewer) blessing = await self.db.get( FeedBlessing, blessing_id, with_for_update=True, ) if ( blessing is None or str(blessing.celebration_id) != str(celebration.id) ): raise NotFoundError("blessing not found") if blessing.acked_at is not None: return blessing, False blessing.acked_at = datetime.now(UTC) await self.db.flush() return blessing, True async def _find_thanks_post( self, celebration_id: str, celebrant_id: str, ) -> FeedPost | None: """The celebrant's 回谢 post for this celebration, if one exists. Soft-deleted rows count. The post is the idempotency marker for ``thank_all_blessings``, and if deleting it re-armed the flow, a celebrant could post-delete-post and re-fan-out an @-mention DM to every well-wisher each time. This is a LOCKING read. Under MySQL's REPEATABLE READ a plain SELECT answers from the transaction's snapshot, and that snapshot is taken at the request's FIRST read — which is ``get_current_user``, long before this guard runs. So a racing request could hold the celebration row lock, see the other request's committed post as *not yet existing*, and insert a second one. Verified in review: two posts, two rounds of @-mention pushes to every well-wisher. A locking read bypasses the snapshot and sees the latest committed row. """ return (await self.db.execute( select(FeedPost) .where( FeedPost.celebration_id == celebration_id, FeedPost.post_type == POST_TYPE_USER, FeedPost.author_person_id == str(celebrant_id), ) .order_by(FeedPost.created_at.asc()) .limit(1) .with_for_update() )).scalar_one_or_none() async def thank_all_blessings( self, *, celebration_id: str, viewer: IdentityPerson, body: str, ) -> tuple[FeedPost, int, list[str]]: """Acknowledge every blessing at once + publish one 回谢 post. Returns ``(post, acked_count, mentioned_person_ids)``. The caller commits and then runs the ordinary ``notify_mentions_on_post`` chain — every well-wisher hears about the thank-you through the mention channel they already have, with no second notification mechanism to keep in sync. Idempotency is decided HERE, off the stored post (see ``_find_thanks_post``), not off a disabled button: a second call raises ``CELEBRATION_ALREADY_THANKED`` and publishes nothing. """ # Locked: the guard below is a read-then-write and two concurrent # requests otherwise both pass it (verified in review — two posts, two # rounds of mention pushes to everyone who sent a blessing). celebration = await self._require_celebrant( celebration_id, viewer, lock=True, ) prior = await self._find_thanks_post(celebration.id, str(viewer.id)) if prior is not None: raise ConflictError( "you've already thanked everyone for this celebration", code=ERR_CELEBRATION_ALREADY_THANKED, ) blessings = list((await self.db.execute( select(FeedBlessing) .where(FeedBlessing.celebration_id == celebration.id) .order_by(FeedBlessing.created_at.asc()) )).scalars().all()) if not blessings: raise ConflictError( "nobody has sent a blessing yet", code=ERR_CELEBRATION_NO_BLESSINGS, ) # Recipients = distinct senders, earliest blessing first, self # excluded (``send_blessing`` already forbids self-blessing, but the # mention writer must not depend on that staying true). recipients = [ pid for pid in dict.fromkeys( str(b.sender_person_id) for b in blessings ) if pid != str(viewer.id) ] # Same cap as every other mention writer. Note the consequence: a # celebrant with more than MAX_MENTIONS_PER_POST well-wishers thanks # everyone on the page (all blessings flip to acked) but only the # first N get a notification — the cap is what keeps a single post # from fanning out an unbounded number of Feishu DMs, and # ``notify_mentions_on_post`` enforces it independently anyway. recipients = recipients[:MAX_MENTIONS_PER_POST] now = datetime.now(UTC) post = FeedPost( author_person_id=str(viewer.id), post_type=POST_TYPE_USER, visibility=VISIBILITY_PUBLIC, category=POST_CATEGORY_GENERAL, body=body, # The link back to the celebration is what makes this post the # durable idempotency marker (and lets the FE deep-link the # thank-you back to the celebration page). ``post_type`` stays # ``user`` — these are the celebrant's own words. celebration_id=celebration.id, department_snapshot=_resolve_dept(viewer), created_at=now, updated_at=now, ) self.db.add(post) await self.db.flush() for pid in recipients: self.db.add(FeedMention( mentioned_person_id=pid, post_id=post.id, author_person_id=str(viewer.id), )) acked_count = 0 for b in blessings: if b.acked_at is None: b.acked_at = now acked_count += 1 await self.db.flush() return post, acked_count, recipients # ──────────────────────────────────────────────────────────── # Announcements # ──────────────────────────────────────────────────────────── async def list_active_announcements( self, *, now: datetime, limit: int = 8, ) -> list[FeedAnnouncement]: q = ( select(FeedAnnouncement) .where( FeedAnnouncement.deleted_at.is_(None), FeedAnnouncement.status == ANNOUNCEMENT_STATUS_PUBLISHED, or_( FeedAnnouncement.publish_at.is_(None), FeedAnnouncement.publish_at <= now, ), or_( FeedAnnouncement.expire_at.is_(None), FeedAnnouncement.expire_at > now, ), ) .order_by( desc(FeedAnnouncement.is_pinned), desc(FeedAnnouncement.publish_at), desc(FeedAnnouncement.created_at), ) .limit(limit) ) return list((await self.db.execute(q)).scalars().all()) async def list_admin_announcements( self, *, include_archived: bool = False, search: str | None = None, limit: int = 200, ) -> list[FeedAnnouncement]: q = select(FeedAnnouncement).where( FeedAnnouncement.deleted_at.is_(None), ) if not include_archived: q = q.where(FeedAnnouncement.status != ANNOUNCEMENT_STATUS_ARCHIVED) if search: # LIKE with ESCAPE so % _ \ in the needle match literally # (same convention as ``search_posts`` above). ``ilike`` # keeps the match case-insensitive regardless of column # collation. pattern = "%" + search.replace("\\", "\\\\")\ .replace("%", "\\%").replace("_", "\\_") + "%" q = q.where( or_( FeedAnnouncement.title.ilike(pattern, escape="\\"), FeedAnnouncement.category.ilike(pattern, escape="\\"), FeedAnnouncement.body_markdown.ilike(pattern, escape="\\"), ) ) q = q.order_by( desc(FeedAnnouncement.is_pinned), desc(FeedAnnouncement.created_at), ).limit(max(1, limit)) return list((await self.db.execute(q)).scalars().all()) async def create_announcement( self, *, author: IdentityPerson, title: str, body_markdown: str, category: str, color: str, is_pinned: bool, publish_at: datetime | None, expire_at: datetime | None, status: str, attachments_json: list[dict[str, Any]] | None, ) -> FeedAnnouncement: if not _is_admin(author): raise ForbiddenError("not authorized to publish announcements") # Same key-shape gate as user posts. Announcements are admin-only # and are not presigned by this service, so there is no # escalation here — but a stored key that no purpose recognises # is a latent one, and keeping both write paths on one rule is # how the next attachment surface inherits the check for free. _validate_attachment_keys(attachments_json or [], owner_id=None) # Python-side timestamps so the router's serializer can read the # values without a lazy-load round-trip (which in our async # session raises MissingGreenlet and surfaces as a 500 AFTER # the row was already committed — exactly the bug pattern of # ``add_comment`` / ``create_user_post``). now = datetime.now(UTC) ann = FeedAnnouncement( title=title, body_markdown=body_markdown, category=category, color=color, is_pinned=is_pinned, publish_at=publish_at, expire_at=expire_at, status=status, attachments_json=attachments_json, author_person_id=str(author.id), created_at=now, updated_at=now, ) self.db.add(ann) await self.db.flush() return ann async def update_announcement( self, *, actor: IdentityPerson, announcement_id: str, patch: dict[str, Any], ) -> FeedAnnouncement: if not _is_admin(actor): raise ForbiddenError("not authorized") ann = await self.db.get(FeedAnnouncement, announcement_id) if ann is None or ann.deleted_at is not None: raise NotFoundError("announcement not found") for k, v in patch.items(): if v is None: continue setattr(ann, k, v) # Mirror server-side onupdate=NOW() Python-side so the response # serializer reads a fresh value without a refresh round-trip. ann.updated_at = datetime.now(UTC) await self.db.flush() return ann async def soft_delete_announcement( self, *, actor: IdentityPerson, announcement_id: str, ) -> None: if not _is_admin(actor): raise ForbiddenError("not authorized") ann = await self.db.get(FeedAnnouncement, announcement_id) if ann is None or ann.deleted_at is not None: raise NotFoundError("announcement not found") ann.deleted_at = datetime.now(UTC) await self.db.flush() # ──────────────────────────────────────────────────────────── # Preferences # ──────────────────────────────────────────────────────────── async def get_preferences( self, *, person: IdentityPerson, ) -> dict[str, Any]: rows = (await self.db.execute( select(IdentityPersonPreference) .where(IdentityPersonPreference.person_id == str(person.id)) )).scalars().all() store = {r.pref_key: r.pref_value for r in rows} return { "birthday": person.birthday, "birthday_visibility": store.get( PREF_BIRTHDAY_VISIBILITY, DEFAULT_PREFERENCES[PREF_BIRTHDAY_VISIBILITY], ), "notify_comment_feishu": store.get( "feed.notify_comment_feishu", DEFAULT_PREFERENCES["feed.notify_comment_feishu"], ), "notify_like_aggregate": store.get( "feed.notify_like_aggregate", DEFAULT_PREFERENCES["feed.notify_like_aggregate"], ), "theme_mode": store.get( PREF_THEME_MODE, DEFAULT_PREFERENCES[PREF_THEME_MODE], ), "festive_effects_enabled": store.get( PREF_FESTIVE_EFFECTS, DEFAULT_PREFERENCES[PREF_FESTIVE_EFFECTS], ), "wrapped_push_opt_out": store.get( PREF_WRAPPED_PUSH_OPT_OUT, DEFAULT_PREFERENCES[PREF_WRAPPED_PUSH_OPT_OUT], ), } async def update_preferences( self, *, person: IdentityPerson, patch: dict[str, Any], ) -> dict[str, Any]: # Birthday lives directly on identity_persons. if "birthday" in patch: bday = patch["birthday"] if isinstance(bday, str): bday = date.fromisoformat(bday) person.birthday = bday # ``theme_mode`` is enum-validated at the Pydantic boundary # (PreferencesUpdate.theme_mode is a Literal) so by the time a # value reaches here it is already one of THEME_MODE_VALUES. # Other keys go into identity_person_preferences. kv_map = { "birthday_visibility": PREF_BIRTHDAY_VISIBILITY, "notify_comment_feishu": "feed.notify_comment_feishu", "notify_like_aggregate": "feed.notify_like_aggregate", "theme_mode": PREF_THEME_MODE, "festive_effects_enabled": PREF_FESTIVE_EFFECTS, "wrapped_push_opt_out": PREF_WRAPPED_PUSH_OPT_OUT, } for ui_key, pref_key in kv_map.items(): if ui_key not in patch: continue value = patch[ui_key] if value is None: continue existing = (await self.db.execute( select(IdentityPersonPreference).where( IdentityPersonPreference.person_id == str(person.id), IdentityPersonPreference.pref_key == pref_key, ) )).scalar_one_or_none() if existing is None: self.db.add(IdentityPersonPreference( person_id=str(person.id), pref_key=pref_key, pref_value=value, )) else: existing.pref_value = value await self.db.flush() return await self.get_preferences(person=person) async def get_birthday_visibility( db: AsyncSession, person_id: str, ) -> str: """Standalone helper used by the celebration scanner (avoids a full FeedService instance for one read). """ row = (await db.execute( select(IdentityPersonPreference.pref_value).where( IdentityPersonPreference.person_id == person_id, IdentityPersonPreference.pref_key == PREF_BIRTHDAY_VISIBILITY, ) )).scalar_one_or_none() if row is None: return DEFAULT_PREFERENCES[PREF_BIRTHDAY_VISIBILITY] return row if isinstance(row, str) else BIRTHDAY_VISIBILITY_HIDDEN