feat: notifier beta

mission/admin.py

@@ -0,0 +1,44 @@
+from django.contrib import admin
+
+from mission.models import Mission, MissionCategory, MissionParticipant, MissionReply
+
+
+@admin.register(MissionCategory)
+class MissionCategoryAdmin(admin.ModelAdmin):
+    list_display = ["id", "merchant", "name", "created_at"]
+    list_filter = ["merchant", "created_at"]
+    search_fields = ["name"]
+    readonly_fields = ["created_at", "updated_at"]
+
+
+@admin.register(Mission)
+class MissionAdmin(admin.ModelAdmin):
+    list_display = [
+        "id",
+        "merchant",
+        "category",
+        "creator",
+        "is_urgent",
+        "is_completed",
+        "is_cancelled",
+        "created_at",
+    ]
+    list_filter = ["merchant", "category", "is_urgent", "is_completed", "is_cancelled", "created_at"]
+    search_fields = ["description", "creator__name"]
+    readonly_fields = ["created_at", "updated_at"]
+
+
+@admin.register(MissionParticipant)
+class MissionParticipantAdmin(admin.ModelAdmin):
+    list_display = ["id", "merchant", "mission", "employee", "created_at"]
+    list_filter = ["merchant", "created_at"]
+    search_fields = ["employee__name"]
+    readonly_fields = ["created_at", "updated_at"]
+
+
+@admin.register(MissionReply)
+class MissionReplyAdmin(admin.ModelAdmin):
+    list_display = ["id", "merchant", "mission", "responder", "ends_task", "is_rejected", "replied_at"]
+    list_filter = ["merchant", "ends_task", "is_rejected", "replied_at"]
+    search_fields = ["content", "responder__name"]
+    readonly_fields = ["created_at", "updated_at"]

mission/apps.py

@@ -0,0 +1,50 @@
+from django.apps import AppConfig
+
+
+class MissionConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'mission'
+    verbose_name = '任务管理'
+
+    def ready(self):
+        from . import handlers
+        from .models import Mission, MissionReply
+        from .signals import (
+            mission_cancelled,
+            mission_completed,
+            mission_created,
+            mission_reopened,
+            mission_replied,
+            mission_reply_rejected,
+        )
+
+        mission_created.connect(
+            handlers.on_mission_created,
+            sender=Mission,
+            dispatch_uid="mission.on_mission_created",
+        )
+        mission_replied.connect(
+            handlers.on_mission_replied,
+            sender=MissionReply,
+            dispatch_uid="mission.on_mission_replied",
+        )
+        mission_completed.connect(
+            handlers.on_mission_completed,
+            sender=Mission,
+            dispatch_uid="mission.on_mission_completed",
+        )
+        mission_reply_rejected.connect(
+            handlers.on_mission_reply_rejected,
+            sender=MissionReply,
+            dispatch_uid="mission.on_mission_reply_rejected",
+        )
+        mission_reopened.connect(
+            handlers.on_mission_reopened,
+            sender=Mission,
+            dispatch_uid="mission.on_mission_reopened",
+        )
+        mission_cancelled.connect(
+            handlers.on_mission_cancelled,
+            sender=Mission,
+            dispatch_uid="mission.on_mission_cancelled",
+        )

mission/handlers.py

@@ -0,0 +1,145 @@
+import logging
+
+from notifier.models import NotificationEventKeyEnum
+from notifier.services import enqueue_notification_event
+
+logger = logging.getLogger(__name__)
+
+
+def _build_mission_payload(mission) -> dict:
+    participant_names = list(
+        mission.get_participants().values_list("employee__name", flat=True)
+    )
+    content_type = getattr(mission.content_type, "model", None)
+    return {
+        "mission_id": mission.id,
+        "merchant_id": mission.merchant_id,
+        "description": mission.description,
+        "category_id": mission.category_id,
+        "category_name": mission.category.name,
+        "is_urgent": mission.is_urgent,
+        "is_completed": mission.is_completed,
+        "is_cancelled": mission.is_cancelled,
+        "creator_id": mission.creator_id,
+        "creator_name": getattr(mission.creator, "name", ""),
+        "participant_names": participant_names,
+        "participant_names_display": "、".join(participant_names) if participant_names else "无",
+        "content_type": content_type or "",
+        "content_id": mission.content_id or "",
+    }
+
+
+def _build_reply_payload(reply) -> dict:
+    return {
+        "reply_id": reply.id,
+        "reply_content": reply.content,
+        "reply_content_short": (reply.content or "")[:100],
+        "responder_id": reply.responder_id,
+        "responder_name": getattr(reply.responder, "name", ""),
+        "ends_task": reply.ends_task,
+        "is_rejected": reply.is_rejected,
+        "replied_at": reply.replied_at.isoformat() if reply.replied_at else "",
+        "rejected_at": reply.rejected_at.isoformat() if reply.rejected_at else "",
+    }
+
+
+def _enqueue(*, event_key: str, merchant_id: int, payload: dict) -> None:
+    task_id = enqueue_notification_event(
+        event_key=event_key,
+        merchant_id=merchant_id,
+        payload=payload,
+    )
+    logger.info(
+        "[mission.handlers] queued notifier event: event_key=%s merchant_id=%s task_id=%s",
+        event_key,
+        merchant_id,
+        task_id,
+    )
+
+
+def on_mission_created(sender, instance, created_by=None, **kwargs):
+    payload = {
+        **_build_mission_payload(instance),
+        "created_by_id": getattr(created_by, "id", None),
+        "created_by_name": getattr(created_by, "name", ""),
+    }
+    _enqueue(
+        event_key=NotificationEventKeyEnum.MISSION_CREATED,
+        merchant_id=instance.merchant_id,
+        payload=payload,
+    )
+
+
+def on_mission_replied(sender, instance, mission=None, responder=None, **kwargs):
+    mission = mission or instance.mission
+    payload = {
+        **_build_mission_payload(mission),
+        **_build_reply_payload(instance),
+        "responder_id": getattr(responder, "id", None),
+        "responder_name": getattr(responder, "name", ""),
+    }
+    _enqueue(
+        event_key=NotificationEventKeyEnum.MISSION_REPLIED,
+        merchant_id=mission.merchant_id,
+        payload=payload,
+    )
+
+
+def on_mission_completed(sender, instance, completed_by=None, reply=None, **kwargs):
+    payload = {
+        **_build_mission_payload(instance),
+        "completed_by_id": getattr(completed_by, "id", None),
+        "completed_by_name": getattr(completed_by, "name", ""),
+    }
+    if reply is not None:
+        payload.update(_build_reply_payload(reply))
+    _enqueue(
+        event_key=NotificationEventKeyEnum.MISSION_COMPLETED,
+        merchant_id=instance.merchant_id,
+        payload=payload,
+    )
+
+
+def on_mission_reply_rejected(sender, instance, mission=None, rejected_by=None, reason=None, **kwargs):
+    mission = mission or instance.mission
+    payload = {
+        **_build_mission_payload(mission),
+        **_build_reply_payload(instance),
+        "rejected_by_id": getattr(rejected_by, "id", None),
+        "rejected_by_name": getattr(rejected_by, "name", ""),
+        "reason": reason or "",
+    }
+    _enqueue(
+        event_key=NotificationEventKeyEnum.MISSION_REPLY_REJECTED,
+        merchant_id=mission.merchant_id,
+        payload=payload,
+    )
+
+
+def on_mission_reopened(sender, instance, reopened_by=None, rejected_reply_ids=None, **kwargs):
+    payload = {
+        **_build_mission_payload(instance),
+        "reopened_by_id": getattr(reopened_by, "id", None),
+        "reopened_by_name": getattr(reopened_by, "name", ""),
+        "rejected_reply_ids": rejected_reply_ids or [],
+        "rejected_reply_ids_display": ", ".join(str(reply_id) for reply_id in (rejected_reply_ids or [])) or "无",
+    }
+    _enqueue(
+        event_key=NotificationEventKeyEnum.MISSION_REOPENED,
+        merchant_id=instance.merchant_id,
+        payload=payload,
+    )
+
+
+def on_mission_cancelled(sender, instance, cancelled_by=None, **kwargs):
+    payload = {
+        **_build_mission_payload(instance),
+        "cancelled_by_id": getattr(cancelled_by, "id", None),
+        "cancelled_by_name": getattr(cancelled_by, "name", ""),
+        "cancelled_at": instance.cancelled_at.isoformat() if instance.cancelled_at else "",
+    }
+    _enqueue(
+        event_key=NotificationEventKeyEnum.MISSION_CANCELLED,
+        merchant_id=instance.merchant_id,
+        payload=payload,
+    )

mission/migrations/0001_initial.py

@@ -0,0 +1,86 @@
+# Generated by Django 5.2.8 on 2026-04-10 02:39
+
+import django.db.models.deletion
+import django.utils.timezone
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ("basic_info", "0025_customer_uniq_customer_merchant_name"),
+        ("contenttypes", "0002_remove_content_type_name"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="Mission",
+            fields=[
+                ("created_at", models.DateTimeField(auto_now_add=True, verbose_name="创建时间")),
+                ("updated_at", models.DateTimeField(auto_now=True, verbose_name="更新时间")),
+                ("id", models.BigAutoField(primary_key=True, serialize=False)),
+                ("description", models.TextField(verbose_name="任务描述")),
+                ("category", models.CharField(choices=[("general", "通用")], db_index=True, default="general", max_length=50, verbose_name="任务类型")),
+                ("is_urgent", models.BooleanField(db_index=True, default=False, verbose_name="是否紧急")),
+                ("is_completed", models.BooleanField(db_index=True, default=False, verbose_name="是否完成")),
+                ("is_cancelled", models.BooleanField(db_index=True, default=False, verbose_name="是否取消")),
+                ("cancelled_at", models.DateTimeField(blank=True, null=True, verbose_name="取消时间")),
+                ("content_id", models.PositiveBigIntegerField(blank=True, db_index=True, null=True, verbose_name="关联对象ID")),
+                ("cancelled_by", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name="cancelled_missions", to="basic_info.employee", verbose_name="取消人")),
+                ("content_type", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="missions", to="contenttypes.contenttype", verbose_name="关联对象类型")),
+                ("creator", models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name="created_missions", to="basic_info.employee", verbose_name="任务创建者")),
+            ],
+            options={
+                "verbose_name": "任务",
+                "verbose_name_plural": "任务",
+            },
+        ),
+        migrations.CreateModel(
+            name="MissionParticipant",
+            fields=[
+                ("created_at", models.DateTimeField(auto_now_add=True, verbose_name="创建时间")),
+                ("updated_at", models.DateTimeField(auto_now=True, verbose_name="更新时间")),
+                ("id", models.BigAutoField(primary_key=True, serialize=False)),
+                ("employee", models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name="mission_participations", to="basic_info.employee", verbose_name="参与者")),
+                ("mission", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="participants", to="mission.mission", verbose_name="任务")),
+            ],
+            options={
+                "verbose_name": "任务参与者",
+                "verbose_name_plural": "任务参与者",
+            },
+        ),
+        migrations.CreateModel(
+            name="MissionReply",
+            fields=[
+                ("created_at", models.DateTimeField(auto_now_add=True, verbose_name="创建时间")),
+                ("updated_at", models.DateTimeField(auto_now=True, verbose_name="更新时间")),
+                ("id", models.BigAutoField(primary_key=True, serialize=False)),
+                ("content", models.TextField(verbose_name="回应内容")),
+                ("replied_at", models.DateTimeField(db_index=True, default=django.utils.timezone.now, verbose_name="回应时间")),
+                ("ends_task", models.BooleanField(db_index=True, default=False, verbose_name="是否结束任务")),
+                ("is_rejected", models.BooleanField(db_index=True, default=False, verbose_name="是否被撤销")),
+                ("rejected_at", models.DateTimeField(blank=True, null=True, verbose_name="撤销时间")),
+                ("mission", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="replies", to="mission.mission", verbose_name="任务")),
+                ("responder", models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name="mission_replies", to="basic_info.employee", verbose_name="回应者")),
+            ],
+            options={
+                "verbose_name": "任务回应",
+                "verbose_name_plural": "任务回应",
+                "ordering": ["replied_at", "id"],
+            },
+        ),
+        migrations.AddIndex(
+            model_name="mission",
+            index=models.Index(fields=["content_type", "content_id"], name="mission_mis_content_be6081_idx"),
+        ),
+        migrations.AddIndex(
+            model_name="mission",
+            index=models.Index(fields=["is_completed", "is_cancelled"], name="mission_mis_is_comp_299039_idx"),
+        ),
+        migrations.AddConstraint(
+            model_name="missionparticipant",
+            constraint=models.UniqueConstraint(fields=("mission", "employee"), name="unique_mission_participant"),
+        ),
+    ]

mission/migrations/0002_add_merchant_and_reply_rejected_by.py

@@ -0,0 +1,125 @@
+# Generated by Django 5.2.8 on 2026-04-10 03:36
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+def populate_merchant_fields(apps, schema_editor):
+    Mission = apps.get_model("mission", "Mission")
+    MissionParticipant = apps.get_model("mission", "MissionParticipant")
+    MissionReply = apps.get_model("mission", "MissionReply")
+
+    for mission in Mission.objects.select_related("creator").filter(merchant__isnull=True):
+        mission.merchant_id = mission.creator.merchant_id
+        mission.save(update_fields=["merchant"])
+
+    for participant in MissionParticipant.objects.select_related("mission").filter(merchant__isnull=True):
+        participant.merchant_id = participant.mission.merchant_id
+        participant.save(update_fields=["merchant"])
+
+    for reply in MissionReply.objects.select_related("mission").filter(merchant__isnull=True):
+        reply.merchant_id = reply.mission.merchant_id
+        reply.save(update_fields=["merchant"])
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("basic_info", "0025_customer_uniq_customer_merchant_name"),
+        ("mission", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.RemoveIndex(
+            model_name="mission",
+            name="mission_mis_content_be6081_idx",
+        ),
+        migrations.RemoveIndex(
+            model_name="mission",
+            name="mission_mis_is_comp_299039_idx",
+        ),
+        migrations.AddField(
+            model_name="mission",
+            name="merchant",
+            field=models.ForeignKey(
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="missions",
+                to="basic_info.merchant",
+                verbose_name="所属商户",
+            ),
+        ),
+        migrations.AddField(
+            model_name="missionparticipant",
+            name="merchant",
+            field=models.ForeignKey(
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="mission_participants",
+                to="basic_info.merchant",
+                verbose_name="所属商户",
+            ),
+        ),
+        migrations.AddField(
+            model_name="missionreply",
+            name="merchant",
+            field=models.ForeignKey(
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="mission_replies",
+                to="basic_info.merchant",
+                verbose_name="所属商户",
+            ),
+        ),
+        migrations.AddField(
+            model_name="missionreply",
+            name="rejected_by",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="rejected_mission_replies",
+                to="basic_info.employee",
+                verbose_name="撤销人",
+            ),
+        ),
+        migrations.RunPython(populate_merchant_fields, migrations.RunPython.noop),
+        migrations.AlterField(
+            model_name="mission",
+            name="merchant",
+            field=models.ForeignKey(
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="missions",
+                to="basic_info.merchant",
+                verbose_name="所属商户",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="missionparticipant",
+            name="merchant",
+            field=models.ForeignKey(
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="mission_participants",
+                to="basic_info.merchant",
+                verbose_name="所属商户",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="missionreply",
+            name="merchant",
+            field=models.ForeignKey(
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="mission_replies",
+                to="basic_info.merchant",
+                verbose_name="所属商户",
+            ),
+        ),
+        migrations.AddIndex(
+            model_name="mission",
+            index=models.Index(fields=["merchant", "content_type", "content_id"], name="mission_mis_merchan_af589f_idx"),
+        ),
+        migrations.AddIndex(
+            model_name="mission",
+            index=models.Index(fields=["merchant", "is_completed", "is_cancelled"], name="mission_mis_merchan_6d6d37_idx"),
+        ),
+    ]

mission/migrations/0003_alter_mission_options.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.2.8 on 2026-04-10 04:40
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('mission', '0002_add_merchant_and_reply_rejected_by'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='mission',
+            options={'permissions': [('reopen_mission', 'Can reopen mission'), ('reject_mission_reply', 'Can reject mission reply')], 'verbose_name': '任务', 'verbose_name_plural': '任务'},
+        ),
+    ]

mission/migrations/0004_missioncategory_refactor.py

@@ -0,0 +1,101 @@
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+LEGACY_CATEGORY_NAME_MAP = {
+    "general": "通用",
+}
+
+
+def forwards(apps, schema_editor):
+    Mission = apps.get_model("mission", "Mission")
+    MissionCategory = apps.get_model("mission", "MissionCategory")
+    db_alias = schema_editor.connection.alias
+    category_cache = {}
+
+    for mission in Mission.objects.using(db_alias).all().only("id", "merchant_id", "category").iterator():
+        legacy_value = (mission.category or "").strip() or "general"
+        category_name = LEGACY_CATEGORY_NAME_MAP.get(legacy_value, legacy_value)
+        cache_key = (mission.merchant_id, category_name)
+
+        category = category_cache.get(cache_key)
+        if category is None:
+            category, _ = MissionCategory.objects.using(db_alias).get_or_create(
+                merchant_id=mission.merchant_id,
+                name=category_name,
+            )
+            category_cache[cache_key] = category
+
+        Mission.objects.using(db_alias).filter(id=mission.id).update(category_ref_id=category.id)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("basic_info", "0025_customer_uniq_customer_merchant_name"),
+        ("mission", "0003_alter_mission_options"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="MissionCategory",
+            fields=[
+                ("created_at", models.DateTimeField(auto_now_add=True, verbose_name="创建时间")),
+                ("updated_at", models.DateTimeField(auto_now=True, verbose_name="更新时间")),
+                ("id", models.BigAutoField(primary_key=True, serialize=False)),
+                ("name", models.CharField(max_length=50, verbose_name="分类名称")),
+                (
+                    "merchant",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.PROTECT,
+                        related_name="mission_categories",
+                        to="basic_info.merchant",
+                        verbose_name="所属商户",
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "任务分类",
+                "verbose_name_plural": "任务分类",
+                "ordering": ["id"],
+                "constraints": [
+                    models.UniqueConstraint(
+                        fields=("merchant", "name"),
+                        name="unique_mission_category_name_per_merchant",
+                    ),
+                ],
+            },
+        ),
+        migrations.AddField(
+            model_name="mission",
+            name="category_ref",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="+",
+                to="mission.missioncategory",
+                verbose_name="任务类型",
+            ),
+        ),
+        migrations.RunPython(forwards, migrations.RunPython.noop),
+        migrations.RemoveField(
+            model_name="mission",
+            name="category",
+        ),
+        migrations.RenameField(
+            model_name="mission",
+            old_name="category_ref",
+            new_name="category",
+        ),
+        migrations.AlterField(
+            model_name="mission",
+            name="category",
+            field=models.ForeignKey(
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="missions",
+                to="mission.missioncategory",
+                verbose_name="任务类型",
+            ),
+        ),
+    ]

mission/models.py

@@ -0,0 +1,189 @@
+from django.contrib.contenttypes.fields import GenericForeignKey
+from django.contrib.contenttypes.models import ContentType
+from django.db import models
+from django.utils import timezone
+
+from flower.common import ModelBase
+
+
+class MissionCategory(ModelBase):
+    """任务分类字典，按商户隔离。"""
+
+    id = models.BigAutoField(primary_key=True)
+    merchant = models.ForeignKey(
+        "basic_info.Merchant",
+        on_delete=models.PROTECT,
+        related_name="mission_categories",
+        verbose_name="所属商户",
+    )
+    name = models.CharField(max_length=50, verbose_name="分类名称")
+
+    def __str__(self):
+        return self.name
+
+    class Meta:
+        verbose_name = "任务分类"
+        verbose_name_plural = "任务分类"
+        ordering = ["id"]
+        constraints = [
+            models.UniqueConstraint(fields=["merchant", "name"], name="unique_mission_category_name_per_merchant"),
+        ]
+
+
+class Mission(ModelBase):
+    """中立任务，可选关联到系统中的任意业务对象。"""
+
+    id = models.BigAutoField(primary_key=True)
+    merchant = models.ForeignKey(
+        "basic_info.Merchant",
+        on_delete=models.PROTECT,
+        related_name="missions",
+        verbose_name="所属商户",
+    )
+    description = models.TextField(verbose_name="任务描述")
+    category = models.ForeignKey(
+        MissionCategory,
+        on_delete=models.PROTECT,
+        related_name="missions",
+        verbose_name="任务类型",
+    )
+    is_urgent = models.BooleanField(default=False, db_index=True, verbose_name="是否紧急")
+    is_completed = models.BooleanField(default=False, db_index=True, verbose_name="是否完成")
+    is_cancelled = models.BooleanField(default=False, db_index=True, verbose_name="是否取消")
+    cancelled_at = models.DateTimeField(null=True, blank=True, verbose_name="取消时间")
+    creator = models.ForeignKey(
+        "basic_info.Employee",
+        on_delete=models.PROTECT,
+        related_name="created_missions",
+        verbose_name="任务创建者",
+    )
+    cancelled_by = models.ForeignKey(
+        "basic_info.Employee",
+        on_delete=models.PROTECT,
+        related_name="cancelled_missions",
+        null=True,
+        blank=True,
+        verbose_name="取消人",
+    )
+    content_type = models.ForeignKey(
+        ContentType,
+        on_delete=models.SET_NULL,
+        related_name="missions",
+        null=True,
+        blank=True,
+        verbose_name="关联对象类型",
+    )
+    content_id = models.PositiveBigIntegerField(null=True, blank=True, db_index=True, verbose_name="关联对象ID")
+    content_object = GenericForeignKey("content_type", "content_id")
+
+    @property
+    def has_ending_reply(self) -> bool:
+        return self.replies.filter(ends_task=True, is_rejected=False).exists()
+
+    @property
+    def can_reply(self) -> bool:
+        return not self.is_cancelled and not self.has_ending_reply
+
+    def get_participants(self):
+        return self.participants.select_related("employee")
+
+    def filter_participants(self, **filters):
+        return self.get_participants().filter(**filters)
+
+    @property
+    def category_name(self) -> str:
+        return self.category.name
+
+    def __str__(self):
+        return f"Mission #{self.id}"
+
+    class Meta:
+        verbose_name = "任务"
+        verbose_name_plural = "任务"
+        permissions = [
+            ("reopen_mission", "Can reopen mission"),
+            ("reject_mission_reply", "Can reject mission reply"),
+        ]
+        indexes = [
+            models.Index(fields=["merchant", "content_type", "content_id"]),
+            models.Index(fields=["merchant", "is_completed", "is_cancelled"]),
+        ]
+
+
+class MissionParticipant(ModelBase):
+    """任务参与员工。"""
+
+    id = models.BigAutoField(primary_key=True)
+    merchant = models.ForeignKey(
+        "basic_info.Merchant",
+        on_delete=models.PROTECT,
+        related_name="mission_participants",
+        verbose_name="所属商户",
+    )
+    mission = models.ForeignKey(
+        Mission,
+        on_delete=models.CASCADE,
+        related_name="participants",
+        verbose_name="任务",
+    )
+    employee = models.ForeignKey(
+        "basic_info.Employee",
+        on_delete=models.PROTECT,
+        related_name="mission_participations",
+        verbose_name="参与者",
+    )
+
+    def __str__(self):
+        return f"{self.mission_id} - {self.employee_id}"
+
+    class Meta:
+        verbose_name = "任务参与者"
+        verbose_name_plural = "任务参与者"
+        constraints = [
+            models.UniqueConstraint(fields=["mission", "employee"], name="unique_mission_participant"),
+        ]
+
+
+class MissionReply(ModelBase):
+    """任务回应记录。"""
+
+    id = models.BigAutoField(primary_key=True)
+    merchant = models.ForeignKey(
+        "basic_info.Merchant",
+        on_delete=models.PROTECT,
+        related_name="mission_replies",
+        verbose_name="所属商户",
+    )
+    mission = models.ForeignKey(
+        Mission,
+        on_delete=models.CASCADE,
+        related_name="replies",
+        verbose_name="任务",
+    )
+    responder = models.ForeignKey(
+        "basic_info.Employee",
+        on_delete=models.PROTECT,
+        related_name="mission_replies",
+        verbose_name="回应者",
+    )
+    content = models.TextField(verbose_name="回应内容")
+    replied_at = models.DateTimeField(default=timezone.now, db_index=True, verbose_name="回应时间")
+    ends_task = models.BooleanField(default=False, db_index=True, verbose_name="是否结束任务")
+    is_rejected = models.BooleanField(default=False, db_index=True, verbose_name="是否被撤销")
+    rejected_by = models.ForeignKey(
+        "basic_info.Employee",
+        on_delete=models.PROTECT,
+        related_name="rejected_mission_replies",
+        null=True,
+        blank=True,
+        verbose_name="撤销人",
+    )
+    rejected_at = models.DateTimeField(null=True, blank=True, verbose_name="撤销时间")
+
+    def __str__(self):
+        return f"MissionReply #{self.id}"
+
+    class Meta:
+        verbose_name = "任务回应"
+        verbose_name_plural = "任务回应"
+        ordering = ["replied_at", "id"]

mission/services.py

@@ -0,0 +1,308 @@
+import logging
+
+from django.db import transaction
+from django.utils import timezone
+
+from mission.models import Mission, MissionCategory, MissionParticipant, MissionReply
+from mission.signals import (
+    mission_cancelled,
+    mission_completed,
+    mission_created,
+    mission_reopened,
+    mission_replied,
+    mission_reply_rejected,
+)
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_MISSION_CATEGORY_NAME = "通用"
+
+
+def _send_signal_on_commit(signal, *, sender, **payload) -> None:
+    def _send():
+        try:
+            signal.send(sender=sender, **payload)
+        except Exception:
+            logger.exception("[mission.services] 触发 mission signal 失败（已忽略）")
+
+    transaction.on_commit(_send)
+
+
+def _assert_employee_belongs_to_mission(employee, mission: Mission, role: str) -> None:
+    if employee is None:
+        raise ValueError(f"{role}不能为空")
+    if employee.merchant_id != mission.merchant_id:
+        raise ValueError(f"{role}不属于任务所属商户")
+
+
+def _get_default_mission_category(*, merchant) -> MissionCategory:
+    category, _ = MissionCategory.objects.get_or_create(
+        merchant=merchant,
+        name=DEFAULT_MISSION_CATEGORY_NAME,
+    )
+    return category
+
+
+def _assert_category_belongs_to_merchant(category: MissionCategory | None, *, merchant) -> None:
+    if category is None:
+        return
+    if category.merchant_id != merchant.id:
+        raise ValueError("任务分类不属于当前商户")
+
+
+def _validate_content_object_merchant(*, content_type, content_id, merchant) -> None:
+    if content_type is None and content_id is None:
+        return
+    if content_type is None or content_id is None:
+        raise ValueError("content_type 与 content_id 必须同时提供或同时为空")
+
+    try:
+        content_object = content_type.get_object_for_this_type(pk=content_id)
+    except Exception as exc:
+        raise ValueError("关联业务对象不存在") from exc
+
+    content_object_merchant_id = getattr(content_object, "merchant_id", None)
+    if content_object_merchant_id is not None and content_object_merchant_id != merchant.id:
+        raise ValueError("关联业务对象不属于当前商户")
+
+
+@transaction.atomic
+def set_mission_participants(*, mission: Mission, participant_ids: list[int]) -> Mission:
+    mission = Mission.objects.select_for_update().get(pk=mission.pk)
+    participant_ids = list(dict.fromkeys(participant_ids or []))
+
+    from basic_info.models import Employee
+
+    employees = list(Employee.objects.filter(id__in=participant_ids, merchant=mission.merchant))
+    if len(employees) != len(participant_ids):
+        raise ValueError("参与者不存在或不属于任务所属商户")
+
+    MissionParticipant.objects.filter(mission=mission).exclude(employee_id__in=participant_ids).delete()
+    existing_employee_ids = set(
+        MissionParticipant.objects.filter(mission=mission, employee_id__in=participant_ids)
+        .values_list("employee_id", flat=True)
+    )
+    MissionParticipant.objects.bulk_create(
+        [
+            MissionParticipant(merchant=mission.merchant, mission=mission, employee=employee)
+            for employee in employees
+            if employee.id not in existing_employee_ids
+        ]
+    )
+    return mission
+
+
+@transaction.atomic
+def create_mission(
+    *,
+    creator,
+    description: str,
+    category: MissionCategory | None = None,
+    content_type=None,
+    content_id: int | None = None,
+    participant_ids: list[int] | None = None,
+) -> Mission:
+    if creator is None:
+        raise ValueError("任务创建者不能为空")
+    merchant = creator.merchant
+    _validate_content_object_merchant(content_type=content_type, content_id=content_id, merchant=merchant)
+    _assert_category_belongs_to_merchant(category, merchant=merchant)
+
+    mission = Mission.objects.create(
+        merchant=merchant,
+        creator=creator,
+        description=description,
+        category=category or _get_default_mission_category(merchant=merchant),
+        content_type=content_type,
+        content_id=content_id,
+    )
+    if participant_ids is not None:
+        set_mission_participants(mission=mission, participant_ids=participant_ids)
+    _send_signal_on_commit(
+        mission_created,
+        sender=Mission,
+        instance=mission,
+        created_by=creator,
+    )
+    return mission
+
+
+@transaction.atomic
+def update_mission(
+    *,
+    mission: Mission,
+    updated_by,
+    description: str | None = None,
+    category: MissionCategory | None = None,
+    content_type=None,
+    content_id: int | None = None,
+    update_content_object: bool = False,
+    participant_ids: list[int] | None = None,
+) -> Mission:
+    mission = Mission.objects.select_for_update().get(pk=mission.pk)
+    _assert_employee_belongs_to_mission(updated_by, mission, "更新任务的员工")
+
+    update_fields = []
+    if description is not None:
+        mission.description = description
+        update_fields.append("description")
+    if category is not None:
+        _assert_category_belongs_to_merchant(category, merchant=mission.merchant)
+        mission.category = category
+        update_fields.append("category")
+    if update_content_object:
+        _validate_content_object_merchant(
+            content_type=content_type,
+            content_id=content_id,
+            merchant=mission.merchant,
+        )
+        mission.content_type = content_type
+        mission.content_id = content_id
+        update_fields.extend(["content_type", "content_id"])
+
+    if update_fields:
+        mission.save(update_fields=[*update_fields, "updated_at"])
+    if participant_ids is not None:
+        set_mission_participants(mission=mission, participant_ids=participant_ids)
+    return mission
+
+
+@transaction.atomic
+def create_mission_reply(
+    *,
+    mission: Mission,
+    responder,
+    content: str,
+    ends_task: bool = False,
+) -> MissionReply:
+    mission = Mission.objects.select_for_update().get(pk=mission.pk)
+    _assert_employee_belongs_to_mission(responder, mission, "回应者")
+    if not mission.can_reply:
+        raise ValueError("任务当前不允许继续回应")
+
+    reply = MissionReply.objects.create(
+        merchant=mission.merchant,
+        mission=mission,
+        responder=responder,
+        content=content,
+        ends_task=ends_task,
+    )
+    if ends_task and not mission.is_completed:
+        mission.is_completed = True
+        mission.save(update_fields=["is_completed", "updated_at"])
+        _send_signal_on_commit(
+            mission_completed,
+            sender=Mission,
+            instance=mission,
+            completed_by=responder,
+            reply=reply,
+        )
+    _send_signal_on_commit(
+        mission_replied,
+        sender=MissionReply,
+        instance=reply,
+        mission=mission,
+        responder=responder,
+    )
+    return reply
+
+
+@transaction.atomic
+def reopen_mission(*, mission: Mission, reopened_by) -> Mission:
+    mission = Mission.objects.select_for_update().get(pk=mission.pk)
+    _assert_employee_belongs_to_mission(reopened_by, mission, "重新打开任务的员工")
+    ending_replies = mission.replies.select_for_update().filter(ends_task=True, is_rejected=False)
+    if mission.is_cancelled:
+        raise ValueError("已取消任务不能重新打开")
+    if not mission.is_completed:
+        raise ValueError("未完成任务不能重新打开")
+    if not ending_replies.exists():
+        raise ValueError("任务没有可撤销的结束回应")
+
+    rejected_reply_ids = list(ending_replies.values_list("id", flat=True))
+    now = timezone.now()
+    ending_replies.update(ends_task=False, is_rejected=True, rejected_by=reopened_by, rejected_at=now)
+    mission.is_completed = False
+    mission.save(update_fields=["is_completed", "updated_at"])
+    for reply in MissionReply.objects.filter(id__in=rejected_reply_ids):
+        _send_signal_on_commit(
+            mission_reply_rejected,
+            sender=MissionReply,
+            instance=reply,
+            mission=mission,
+            rejected_by=reopened_by,
+            reason="reopen",
+        )
+    _send_signal_on_commit(
+        mission_reopened,
+        sender=Mission,
+        instance=mission,
+        reopened_by=reopened_by,
+        rejected_reply_ids=rejected_reply_ids,
+    )
+    return mission
+
+
+@transaction.atomic
+def reject_reply(*, reply: MissionReply, rejected_by) -> MissionReply:
+    reply = MissionReply.objects.select_for_update().select_related("mission").get(pk=reply.pk)
+    mission = Mission.objects.select_for_update().get(pk=reply.mission_id)
+    _assert_employee_belongs_to_mission(rejected_by, mission, "撤销回应的员工")
+
+    if mission.is_cancelled:
+        raise ValueError("已取消任务不能撤销回应")
+    if reply.is_rejected:
+        raise ValueError("回应已被撤销")
+
+    was_ending_reply = reply.ends_task
+    now = timezone.now()
+    reply.ends_task = False
+    reply.is_rejected = True
+    reply.rejected_by = rejected_by
+    reply.rejected_at = now
+    reply.save(update_fields=["ends_task", "is_rejected", "rejected_by", "rejected_at", "updated_at"])
+
+    has_other_ending_reply = mission.replies.exclude(pk=reply.pk).filter(ends_task=True, is_rejected=False).exists()
+    if was_ending_reply and not has_other_ending_reply and mission.is_completed:
+        mission.is_completed = False
+        mission.save(update_fields=["is_completed", "updated_at"])
+    _send_signal_on_commit(
+        mission_reply_rejected,
+        sender=MissionReply,
+        instance=reply,
+        mission=mission,
+        rejected_by=rejected_by,
+        reason="reject_reply",
+    )
+    return reply
+
+
+@transaction.atomic
+def cancel_mission(*, mission: Mission, cancelled_by) -> Mission:
+    mission = Mission.objects.select_for_update().get(pk=mission.pk)
+    _assert_employee_belongs_to_mission(cancelled_by, mission, "取消任务的员工")
+
+    if mission.is_cancelled:
+        raise ValueError("任务已取消")
+
+    mission.is_cancelled = True
+    mission.cancelled_by = cancelled_by
+    mission.cancelled_at = timezone.now()
+    mission.save(update_fields=["is_cancelled", "cancelled_by", "cancelled_at", "updated_at"])
+    _send_signal_on_commit(
+        mission_cancelled,
+        sender=Mission,
+        instance=mission,
+        cancelled_by=cancelled_by,
+    )
+    return mission
+
+
+@transaction.atomic
+def set_mission_urgent(*, mission: Mission, updated_by, is_urgent: bool) -> Mission:
+    mission = Mission.objects.select_for_update().get(pk=mission.pk)
+    _assert_employee_belongs_to_mission(updated_by, mission, "更新任务紧急状态的员工")
+
+    mission.is_urgent = is_urgent
+    mission.save(update_fields=["is_urgent", "updated_at"])
+    return mission

mission/signals.py

@@ -0,0 +1,45 @@
+"""
+Mission domain signals.
+
+Signals are emitted by the mission service layer after successful business
+state transitions. Use transaction.on_commit at send sites so handlers only run
+after the database transaction is committed.
+"""
+
+from django.dispatch import Signal
+
+
+# Payload:
+# - instance: Mission
+# - created_by: Employee
+mission_created = Signal()
+
+# Payload:
+# - instance: MissionReply
+# - mission: Mission
+# - responder: Employee
+mission_replied = Signal()
+
+# Payload:
+# - instance: Mission
+# - completed_by: Employee
+# - reply: MissionReply
+mission_completed = Signal()
+
+# Payload:
+# - instance: MissionReply
+# - mission: Mission
+# - rejected_by: Employee
+# - reason: "reject_reply" | "reopen"
+mission_reply_rejected = Signal()
+
+# Payload:
+# - instance: Mission
+# - reopened_by: Employee
+# - rejected_reply_ids: list[int]
+mission_reopened = Signal()
+
+# Payload:
+# - instance: Mission
+# - cancelled_by: Employee
+mission_cancelled = Signal()

mission/tests.py

@@ -0,0 +1,528 @@
+from django.test import TestCase
+from django.contrib.contenttypes.models import ContentType
+from unittest.mock import patch
+
+from basic_info.models import Employee, Merchant, MerchantTypeEnum
+from mission.models import Mission, MissionCategory, MissionParticipant, MissionReply
+from mission.services import (
+    cancel_mission,
+    create_mission,
+    create_mission_reply,
+    reject_reply,
+    reopen_mission,
+    set_mission_participants,
+    set_mission_urgent,
+    update_mission,
+)
+from mission.signals import (
+    mission_cancelled,
+    mission_completed,
+    mission_created,
+    mission_reopened,
+    mission_replied,
+    mission_reply_rejected,
+)
+
+
+class MissionModelTestCase(TestCase):
+    def setUp(self):
+        self.merchant = Merchant.objects.create(name="测试商户", type=MerchantTypeEnum.STORE)
+        self.other_merchant = Merchant.objects.create(name="其他商户", type=MerchantTypeEnum.STORE)
+        self.default_category = MissionCategory.objects.create(merchant=self.merchant, name="通用")
+        self.custom_category = MissionCategory.objects.create(merchant=self.merchant, name="售后")
+        self.other_category = MissionCategory.objects.create(merchant=self.other_merchant, name="通用")
+        self.creator = Employee.objects.create(merchant=self.merchant, name="创建者")
+        self.responder = Employee.objects.create(merchant=self.merchant, name="回应者")
+        self.other_employee = Employee.objects.create(merchant=self.other_merchant, name="外部员工")
+        self.mission = Mission.objects.create(
+            merchant=self.merchant,
+            category=self.default_category,
+            description="测试任务",
+            creator=self.creator,
+        )
+
+    def test_participant_helpers(self):
+        MissionParticipant.objects.create(
+            merchant=self.merchant,
+            mission=self.mission,
+            employee=self.responder,
+        )
+
+        self.assertEqual(self.mission.get_participants().count(), 1)
+        self.assertEqual(
+            self.mission.filter_participants(employee=self.responder).first().employee,
+            self.responder,
+        )
+
+    def test_create_ending_reply_marks_mission_completed(self):
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="完成任务",
+            ends_task=True,
+        )
+
+        self.mission.refresh_from_db()
+        reply.refresh_from_db()
+        self.assertTrue(reply.ends_task)
+        self.assertEqual(reply.merchant, self.merchant)
+        self.assertTrue(self.mission.is_completed)
+        self.assertTrue(self.mission.has_ending_reply)
+        self.assertFalse(self.mission.can_reply)
+
+    def test_reopen_mission_rejects_ending_reply(self):
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="完成任务",
+            ends_task=True,
+        )
+
+        reopen_mission(mission=self.mission, reopened_by=self.creator)
+
+        self.mission.refresh_from_db()
+        reply.refresh_from_db()
+        self.assertFalse(self.mission.is_completed)
+        self.assertFalse(reply.ends_task)
+        self.assertTrue(reply.is_rejected)
+        self.assertEqual(reply.rejected_by, self.creator)
+        self.assertIsNotNone(reply.rejected_at)
+        self.assertTrue(self.mission.can_reply)
+
+    def test_reopen_cancelled_mission_is_rejected(self):
+        create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="完成任务",
+            ends_task=True,
+        )
+        self.mission.is_cancelled = True
+        self.mission.save(update_fields=["is_cancelled", "updated_at"])
+
+        with self.assertRaises(ValueError):
+            reopen_mission(mission=self.mission, reopened_by=self.creator)
+
+    def test_cross_merchant_responder_is_rejected(self):
+        with self.assertRaises(ValueError):
+            create_mission_reply(
+                mission=self.mission,
+                responder=self.other_employee,
+                content="跨商户回应",
+            )
+
+    def test_create_reply_rejects_empty_responder(self):
+        with self.assertRaises(ValueError):
+            create_mission_reply(
+                mission=self.mission,
+                responder=None,
+                content="无回应者",
+            )
+
+    def test_reject_reply_marks_reply_rejected_and_reopens_mission(self):
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="完成任务",
+            ends_task=True,
+        )
+
+        reject_reply(reply=reply, rejected_by=self.creator)
+
+        self.mission.refresh_from_db()
+        reply.refresh_from_db()
+        self.assertFalse(self.mission.is_completed)
+        self.assertFalse(reply.ends_task)
+        self.assertTrue(reply.is_rejected)
+        self.assertEqual(reply.rejected_by, self.creator)
+        self.assertIsNotNone(reply.rejected_at)
+
+    def test_cancel_mission_records_employee_and_time(self):
+        cancel_mission(mission=self.mission, cancelled_by=self.creator)
+
+        self.mission.refresh_from_db()
+        self.assertTrue(self.mission.is_cancelled)
+        self.assertEqual(self.mission.cancelled_by, self.creator)
+        self.assertIsNotNone(self.mission.cancelled_at)
+
+    def test_cross_merchant_cancel_is_rejected(self):
+        with self.assertRaises(ValueError):
+            cancel_mission(mission=self.mission, cancelled_by=self.other_employee)
+
+    def test_create_mission_sets_defaults_and_participants(self):
+        mission = create_mission(
+            creator=self.creator,
+            description="service 创建任务",
+            participant_ids=[self.responder.id, self.responder.id],
+        )
+
+        self.assertEqual(mission.merchant, self.merchant)
+        self.assertEqual(mission.category, self.default_category)
+        self.assertFalse(mission.is_urgent)
+        self.assertEqual(list(mission.participants.values_list("employee_id", flat=True)), [self.responder.id])
+
+    def test_create_mission_rejects_empty_creator(self):
+        with self.assertRaises(ValueError):
+            create_mission(creator=None, description="无创建人")
+
+    def test_create_mission_rejects_partial_content_object(self):
+        with self.assertRaises(ValueError):
+            create_mission(
+                creator=self.creator,
+                description="缺少 content_id",
+                content_type=ContentType.objects.get_for_model(Mission),
+            )
+
+    def test_create_mission_rejects_missing_content_object(self):
+        with self.assertRaises(ValueError):
+            create_mission(
+                creator=self.creator,
+                description="不存在的关联对象",
+                content_type=ContentType.objects.get_for_model(Mission),
+                content_id=999999,
+            )
+
+    def test_create_mission_rejects_cross_merchant_content_object(self):
+        other_mission = Mission.objects.create(
+            merchant=self.other_merchant,
+            category=self.other_category,
+            creator=self.other_employee,
+            description="其他商户任务",
+        )
+
+        with self.assertRaises(ValueError):
+            create_mission(
+                creator=self.creator,
+                description="跨商户关联对象",
+                content_type=ContentType.objects.get_for_model(Mission),
+                content_id=other_mission.id,
+            )
+
+    def test_update_mission_updates_content_object_and_participants(self):
+        related = Mission.objects.create(
+            merchant=self.merchant,
+            category=self.default_category,
+            creator=self.creator,
+            description="同商户关联对象",
+        )
+
+        update_mission(
+            mission=self.mission,
+            updated_by=self.creator,
+            description="更新描述",
+            category=self.custom_category,
+            content_type=ContentType.objects.get_for_model(Mission),
+            content_id=related.id,
+            update_content_object=True,
+            participant_ids=[self.responder.id],
+        )
+
+        self.mission.refresh_from_db()
+        self.assertEqual(self.mission.description, "更新描述")
+        self.assertEqual(self.mission.category, self.custom_category)
+        self.assertEqual(self.mission.content_id, related.id)
+        self.assertEqual(list(self.mission.participants.values_list("employee_id", flat=True)), [self.responder.id])
+
+    def test_create_mission_rejects_cross_merchant_category(self):
+        with self.assertRaises(ValueError):
+            create_mission(
+                creator=self.creator,
+                description="跨商户分类",
+                category=self.other_category,
+            )
+
+    def test_update_mission_rejects_cross_merchant_employee(self):
+        with self.assertRaises(ValueError):
+            update_mission(
+                mission=self.mission,
+                updated_by=self.other_employee,
+                description="非法更新",
+            )
+
+    def test_set_participants_rejects_cross_merchant_employee(self):
+        with self.assertRaises(ValueError):
+            set_mission_participants(
+                mission=self.mission,
+                participant_ids=[self.other_employee.id],
+            )
+
+    def test_update_mission_rejects_cross_merchant_category(self):
+        with self.assertRaises(ValueError):
+            update_mission(
+                mission=self.mission,
+                updated_by=self.creator,
+                category=self.other_category,
+            )
+
+    def test_create_reply_rejects_after_ending_reply(self):
+        create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="结束任务",
+            ends_task=True,
+        )
+
+        with self.assertRaises(ValueError):
+            create_mission_reply(
+                mission=self.mission,
+                responder=self.responder,
+                content="后续回应",
+            )
+
+    def test_reopen_uncompleted_mission_is_rejected(self):
+        with self.assertRaises(ValueError):
+            reopen_mission(mission=self.mission, reopened_by=self.creator)
+
+    def test_reopen_completed_mission_without_ending_reply_is_rejected(self):
+        self.mission.is_completed = True
+        self.mission.save(update_fields=["is_completed", "updated_at"])
+
+        with self.assertRaises(ValueError):
+            reopen_mission(mission=self.mission, reopened_by=self.creator)
+
+    def test_reopen_cross_merchant_employee_is_rejected(self):
+        create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="结束任务",
+            ends_task=True,
+        )
+
+        with self.assertRaises(ValueError):
+            reopen_mission(mission=self.mission, reopened_by=self.other_employee)
+
+    def test_reject_reply_rejects_cancelled_mission(self):
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="普通回应",
+        )
+        self.mission.is_cancelled = True
+        self.mission.save(update_fields=["is_cancelled", "updated_at"])
+
+        with self.assertRaises(ValueError):
+            reject_reply(reply=reply, rejected_by=self.creator)
+
+    def test_reject_reply_rejects_already_rejected_reply(self):
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="普通回应",
+        )
+
+        reject_reply(reply=reply, rejected_by=self.creator)
+
+        with self.assertRaises(ValueError):
+            reject_reply(reply=reply, rejected_by=self.creator)
+
+    def test_reject_reply_rejects_cross_merchant_employee(self):
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="普通回应",
+        )
+
+        with self.assertRaises(ValueError):
+            reject_reply(reply=reply, rejected_by=self.other_employee)
+
+    def test_reject_non_ending_reply_keeps_completed_state(self):
+        self.mission.is_completed = True
+        self.mission.save(update_fields=["is_completed", "updated_at"])
+        reply = MissionReply.objects.create(
+            merchant=self.merchant,
+            mission=self.mission,
+            responder=self.responder,
+            content="普通回应",
+            ends_task=False,
+        )
+
+        reject_reply(reply=reply, rejected_by=self.creator)
+
+        self.mission.refresh_from_db()
+        self.assertTrue(self.mission.is_completed)
+
+    def test_cancel_already_cancelled_mission_is_rejected(self):
+        cancel_mission(mission=self.mission, cancelled_by=self.creator)
+
+        with self.assertRaises(ValueError):
+            cancel_mission(mission=self.mission, cancelled_by=self.creator)
+
+    def test_set_urgent_updates_status_and_rejects_cross_merchant_employee(self):
+        set_mission_urgent(mission=self.mission, updated_by=self.creator, is_urgent=True)
+
+        self.mission.refresh_from_db()
+        self.assertTrue(self.mission.is_urgent)
+
+        with self.assertRaises(ValueError):
+            set_mission_urgent(mission=self.mission, updated_by=self.other_employee, is_urgent=False)
+
+    def test_create_mission_emits_mission_created_signal(self):
+        received = []
+
+        def receiver(sender, instance, created_by=None, **kwargs):
+            received.append((sender, instance.id, created_by.id))
+
+        mission_created.connect(receiver, sender=Mission, dispatch_uid="test_mission_created")
+        try:
+            with self.captureOnCommitCallbacks(execute=True):
+                mission = create_mission(creator=self.creator, description="触发创建信号")
+        finally:
+            mission_created.disconnect(sender=Mission, dispatch_uid="test_mission_created")
+
+        self.assertEqual(received, [(Mission, mission.id, self.creator.id)])
+
+    def test_create_reply_emits_replied_and_completed_signals(self):
+        received = []
+
+        def on_replied(sender, instance, mission=None, responder=None, **kwargs):
+            received.append(("replied", sender, instance.id, mission.id, responder.id))
+
+        def on_completed(sender, instance, completed_by=None, reply=None, **kwargs):
+            received.append(("completed", sender, instance.id, completed_by.id, reply.id))
+
+        mission_replied.connect(on_replied, sender=MissionReply, dispatch_uid="test_mission_replied")
+        mission_completed.connect(on_completed, sender=Mission, dispatch_uid="test_mission_completed")
+        try:
+            with self.captureOnCommitCallbacks(execute=True):
+                reply = create_mission_reply(
+                    mission=self.mission,
+                    responder=self.responder,
+                    content="完成并触发信号",
+                    ends_task=True,
+                )
+        finally:
+            mission_replied.disconnect(sender=MissionReply, dispatch_uid="test_mission_replied")
+            mission_completed.disconnect(sender=Mission, dispatch_uid="test_mission_completed")
+
+        self.assertIn(("completed", Mission, self.mission.id, self.responder.id, reply.id), received)
+        self.assertIn(("replied", MissionReply, reply.id, self.mission.id, self.responder.id), received)
+
+    def test_reject_reply_emits_reply_rejected_signal(self):
+        received = []
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="待撤销回应",
+        )
+
+        def receiver(sender, instance, mission=None, rejected_by=None, reason=None, **kwargs):
+            received.append((sender, instance.id, mission.id, rejected_by.id, reason))
+
+        mission_reply_rejected.connect(receiver, sender=MissionReply, dispatch_uid="test_reply_rejected")
+        try:
+            with self.captureOnCommitCallbacks(execute=True):
+                reject_reply(reply=reply, rejected_by=self.creator)
+        finally:
+            mission_reply_rejected.disconnect(sender=MissionReply, dispatch_uid="test_reply_rejected")
+
+        self.assertEqual(received, [(MissionReply, reply.id, self.mission.id, self.creator.id, "reject_reply")])
+
+    def test_reopen_emits_reopened_and_reply_rejected_signals(self):
+        received = []
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="结束并等待 reopen",
+            ends_task=True,
+        )
+
+        def on_reply_rejected(sender, instance, mission=None, rejected_by=None, reason=None, **kwargs):
+            received.append(("reply_rejected", sender, instance.id, mission.id, rejected_by.id, reason))
+
+        def on_reopened(sender, instance, reopened_by=None, rejected_reply_ids=None, **kwargs):
+            received.append(("reopened", sender, instance.id, reopened_by.id, rejected_reply_ids))
+
+        mission_reply_rejected.connect(on_reply_rejected, sender=MissionReply, dispatch_uid="test_reopen_reply_rejected")
+        mission_reopened.connect(on_reopened, sender=Mission, dispatch_uid="test_mission_reopened")
+        try:
+            with self.captureOnCommitCallbacks(execute=True):
+                reopen_mission(mission=self.mission, reopened_by=self.creator)
+        finally:
+            mission_reply_rejected.disconnect(sender=MissionReply, dispatch_uid="test_reopen_reply_rejected")
+            mission_reopened.disconnect(sender=Mission, dispatch_uid="test_mission_reopened")
+
+        self.assertIn(("reply_rejected", MissionReply, reply.id, self.mission.id, self.creator.id, "reopen"), received)
+        self.assertIn(("reopened", Mission, self.mission.id, self.creator.id, [reply.id]), received)
+
+    def test_cancel_mission_emits_cancelled_signal(self):
+        received = []
+
+        def receiver(sender, instance, cancelled_by=None, **kwargs):
+            received.append((sender, instance.id, cancelled_by.id))
+
+        mission_cancelled.connect(receiver, sender=Mission, dispatch_uid="test_mission_cancelled")
+        try:
+            with self.captureOnCommitCallbacks(execute=True):
+                cancel_mission(mission=self.mission, cancelled_by=self.creator)
+        finally:
+            mission_cancelled.disconnect(sender=Mission, dispatch_uid="test_mission_cancelled")
+
+        self.assertEqual(received, [(Mission, self.mission.id, self.creator.id)])
+
+    @patch("mission.handlers.enqueue_notification_event")
+    def test_mission_created_handler_enqueues_notifier_event(self, mock_enqueue):
+        with self.captureOnCommitCallbacks(execute=True):
+            mission = create_mission(creator=self.creator, description="触发任务创建通知")
+
+        mock_enqueue.assert_called_once()
+        self.assertEqual(mock_enqueue.call_args.kwargs["event_key"], "mission.created")
+        self.assertEqual(mock_enqueue.call_args.kwargs["merchant_id"], self.merchant.id)
+        self.assertEqual(mock_enqueue.call_args.kwargs["payload"]["mission_id"], mission.id)
+
+    @patch("mission.handlers.enqueue_notification_event")
+    def test_create_ending_reply_handler_enqueues_replied_and_completed_notifications(self, mock_enqueue):
+        with self.captureOnCommitCallbacks(execute=True):
+            reply = create_mission_reply(
+                mission=self.mission,
+                responder=self.responder,
+                content="完成任务并通知",
+                ends_task=True,
+            )
+
+        self.assertEqual(mock_enqueue.call_count, 2)
+        event_keys = {call.kwargs["event_key"] for call in mock_enqueue.call_args_list}
+        self.assertEqual(event_keys, {"mission.replied", "mission.completed"})
+        payloads = [call.kwargs["payload"] for call in mock_enqueue.call_args_list]
+        self.assertTrue(any(payload.get("reply_id") == reply.id for payload in payloads))
+
+    @patch("mission.handlers.enqueue_notification_event")
+    def test_reject_reply_handler_enqueues_notification(self, mock_enqueue):
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="待撤销回应",
+        )
+
+        with self.captureOnCommitCallbacks(execute=True):
+            reject_reply(reply=reply, rejected_by=self.creator)
+
+        mock_enqueue.assert_called_once()
+        self.assertEqual(mock_enqueue.call_args.kwargs["event_key"], "mission.reply_rejected")
+        self.assertEqual(mock_enqueue.call_args.kwargs["payload"]["reply_id"], reply.id)
+
+    @patch("mission.handlers.enqueue_notification_event")
+    def test_reopen_mission_handler_enqueues_rejected_and_reopened_notifications(self, mock_enqueue):
+        reply = create_mission_reply(
+            mission=self.mission,
+            responder=self.responder,
+            content="结束任务",
+            ends_task=True,
+        )
+
+        with self.captureOnCommitCallbacks(execute=True):
+            reopen_mission(mission=self.mission, reopened_by=self.creator)
+
+        self.assertEqual(mock_enqueue.call_count, 2)
+        event_keys = {call.kwargs["event_key"] for call in mock_enqueue.call_args_list}
+        self.assertEqual(event_keys, {"mission.reply_rejected", "mission.reopened"})
+        self.assertTrue(
+            any(reply.id in call.kwargs["payload"].get("rejected_reply_ids", []) for call in mock_enqueue.call_args_list)
+        )
+
+    @patch("mission.handlers.enqueue_notification_event")
+    def test_cancel_mission_handler_enqueues_notification(self, mock_enqueue):
+        with self.captureOnCommitCallbacks(execute=True):
+            cancel_mission(mission=self.mission, cancelled_by=self.creator)
+
+        mock_enqueue.assert_called_once()
+        self.assertEqual(mock_enqueue.call_args.kwargs["event_key"], "mission.cancelled")
+        self.assertEqual(mock_enqueue.call_args.kwargs["payload"]["mission_id"], self.mission.id)

mission/views.py

@@ -0,0 +1,3 @@
+from django.shortcuts import render
+
+# Create your views here.