312 lines
12 KiB
Python
312 lines
12 KiB
Python
"""Campaign Engine — lifecycle transitions, rule versioning, and point-granting apply (Phase 7.4)."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, timezone
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.events.publisher import TransactionalEventPublisher
|
|
from app.events.types import LoyaltyEventType
|
|
from app.models.campaigns import CampaignApplication
|
|
from app.models.foundation import Campaign
|
|
from app.models.types import AuditAction, CampaignApplicationStatus
|
|
from app.repositories.campaigns import CampaignApplicationRepository
|
|
from app.repositories.foundation import (
|
|
CampaignRepository,
|
|
MemberRepository,
|
|
PointAccountRepository,
|
|
)
|
|
from app.schemas.campaigns import ApplyCampaignRequest, CampaignRulesUpdateRequest
|
|
from app.schemas.ledger import PointEarnRequest
|
|
from app.services.audit_service import AuditService
|
|
from app.services.point_engine import PointEngineService
|
|
from app.validators.campaigns import (
|
|
compute_grant_amount,
|
|
ensure_campaign_active,
|
|
ensure_campaign_transition,
|
|
ensure_campaign_window,
|
|
ensure_max_applications_per_member,
|
|
ensure_positive_grant,
|
|
ensure_schedule_requires_start_date,
|
|
target_campaign_status,
|
|
)
|
|
from shared.exceptions import AppError, NotFoundError
|
|
from shared.security import CurrentUser
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
_ACTION_EVENTS: dict[str, LoyaltyEventType] = {
|
|
"schedule": LoyaltyEventType.CAMPAIGN_SCHEDULED,
|
|
"activate": LoyaltyEventType.CAMPAIGN_ACTIVATED,
|
|
"pause": LoyaltyEventType.CAMPAIGN_PAUSED,
|
|
"resume": LoyaltyEventType.CAMPAIGN_RESUMED,
|
|
"expire": LoyaltyEventType.CAMPAIGN_EXPIRED,
|
|
}
|
|
|
|
_ACTION_AUDIT: dict[str, AuditAction] = {
|
|
"schedule": AuditAction.SCHEDULE,
|
|
"activate": AuditAction.ACTIVATE,
|
|
"pause": AuditAction.PAUSE,
|
|
"resume": AuditAction.RESUME,
|
|
"expire": AuditAction.EXPIRE,
|
|
}
|
|
|
|
|
|
class CampaignEngineService:
|
|
def __init__(
|
|
self,
|
|
session: AsyncSession,
|
|
publisher: TransactionalEventPublisher | None = None,
|
|
) -> None:
|
|
self.session = session
|
|
self.campaigns = CampaignRepository(session)
|
|
self.applications = CampaignApplicationRepository(session)
|
|
self.members = MemberRepository(session)
|
|
self.accounts = PointAccountRepository(session)
|
|
self.audit = AuditService(session)
|
|
self.publisher = publisher or TransactionalEventPublisher(session)
|
|
self.point_engine = PointEngineService(session, publisher=self.publisher)
|
|
|
|
async def get(self, tenant_id: UUID, campaign_id: UUID) -> Campaign:
|
|
entity = await self.campaigns.get(tenant_id, campaign_id)
|
|
if entity is None:
|
|
raise NotFoundError("کمپین یافت نشد", error_code="campaign_not_found")
|
|
return entity
|
|
|
|
async def _transition(
|
|
self,
|
|
tenant_id: UUID,
|
|
campaign_id: UUID,
|
|
action: str,
|
|
*,
|
|
actor: CurrentUser | None = None,
|
|
) -> Campaign:
|
|
entity = await self.get(tenant_id, campaign_id)
|
|
ensure_campaign_transition(action=action, current=entity.status)
|
|
if action == "schedule":
|
|
ensure_schedule_requires_start_date(entity.starts_on)
|
|
from_status = entity.status
|
|
to_status = target_campaign_status(action)
|
|
actor_id = actor.user_id if actor else None
|
|
entity.status = to_status
|
|
entity.updated_by = actor_id
|
|
await self.audit.record(
|
|
tenant_id=tenant_id,
|
|
entity_type="campaign",
|
|
entity_id=entity.id,
|
|
action=_ACTION_AUDIT[action],
|
|
actor_user_id=actor_id,
|
|
message=(
|
|
f"Campaign {entity.code} {action}: "
|
|
f"{from_status.value} -> {to_status.value}"
|
|
),
|
|
changes={"from_status": from_status.value, "to_status": to_status.value},
|
|
)
|
|
await self.publisher.publish(
|
|
event_type=_ACTION_EVENTS[action],
|
|
aggregate_type="campaign",
|
|
aggregate_id=entity.id,
|
|
tenant_id=tenant_id,
|
|
payload={"code": entity.code, "status": entity.status.value},
|
|
)
|
|
await self.session.commit()
|
|
await self.session.refresh(entity)
|
|
return entity
|
|
|
|
async def schedule(
|
|
self, tenant_id: UUID, campaign_id: UUID, *, actor: CurrentUser | None = None
|
|
) -> Campaign:
|
|
return await self._transition(tenant_id, campaign_id, "schedule", actor=actor)
|
|
|
|
async def activate(
|
|
self, tenant_id: UUID, campaign_id: UUID, *, actor: CurrentUser | None = None
|
|
) -> Campaign:
|
|
return await self._transition(tenant_id, campaign_id, "activate", actor=actor)
|
|
|
|
async def pause(
|
|
self, tenant_id: UUID, campaign_id: UUID, *, actor: CurrentUser | None = None
|
|
) -> Campaign:
|
|
return await self._transition(tenant_id, campaign_id, "pause", actor=actor)
|
|
|
|
async def resume(
|
|
self, tenant_id: UUID, campaign_id: UUID, *, actor: CurrentUser | None = None
|
|
) -> Campaign:
|
|
return await self._transition(tenant_id, campaign_id, "resume", actor=actor)
|
|
|
|
async def expire(
|
|
self, tenant_id: UUID, campaign_id: UUID, *, actor: CurrentUser | None = None
|
|
) -> Campaign:
|
|
return await self._transition(tenant_id, campaign_id, "expire", actor=actor)
|
|
|
|
async def update_rules(
|
|
self,
|
|
tenant_id: UUID,
|
|
campaign_id: UUID,
|
|
body: CampaignRulesUpdateRequest,
|
|
*,
|
|
actor: CurrentUser | None = None,
|
|
) -> Campaign:
|
|
entity = await self.get(tenant_id, campaign_id)
|
|
actor_id = actor.user_id if actor else None
|
|
entity.rules = body.rules
|
|
entity.rule_version += 1
|
|
entity.updated_by = actor_id
|
|
await self.audit.record(
|
|
tenant_id=tenant_id,
|
|
entity_type="campaign",
|
|
entity_id=entity.id,
|
|
action=AuditAction.RULES_UPDATE,
|
|
actor_user_id=actor_id,
|
|
message=f"Campaign {entity.code} rules updated to v{entity.rule_version}",
|
|
changes={"rules": body.rules, "rule_version": entity.rule_version},
|
|
)
|
|
await self.publisher.publish(
|
|
event_type=LoyaltyEventType.CAMPAIGN_RULES_UPDATED,
|
|
aggregate_type="campaign",
|
|
aggregate_id=entity.id,
|
|
tenant_id=tenant_id,
|
|
payload={"code": entity.code, "rule_version": entity.rule_version},
|
|
)
|
|
await self.session.commit()
|
|
await self.session.refresh(entity)
|
|
return entity
|
|
|
|
async def apply(
|
|
self,
|
|
tenant_id: UUID,
|
|
campaign_id: UUID,
|
|
body: ApplyCampaignRequest,
|
|
*,
|
|
actor: CurrentUser | None = None,
|
|
) -> CampaignApplication:
|
|
if body.idempotency_key:
|
|
existing = await self.applications.get_by_idempotency(
|
|
tenant_id, body.idempotency_key
|
|
)
|
|
if existing is not None:
|
|
return existing
|
|
|
|
campaign = await self.get(tenant_id, campaign_id)
|
|
ensure_campaign_active(campaign.status)
|
|
ensure_campaign_window(campaign.starts_on, campaign.ends_on, today=date.today())
|
|
|
|
member = await self.members.get(tenant_id, body.member_id)
|
|
if member is None:
|
|
raise NotFoundError("عضو یافت نشد", error_code="member_not_found")
|
|
if member.program_id != campaign.program_id:
|
|
raise AppError(
|
|
"عضو متعلق به برنامه این کمپین نیست",
|
|
status_code=409,
|
|
error_code="member_program_mismatch",
|
|
)
|
|
|
|
if body.point_account_id is not None:
|
|
account = await self.accounts.get(tenant_id, body.point_account_id)
|
|
if account is None or account.member_id != member.id:
|
|
raise NotFoundError(
|
|
"حساب امتیاز یافت نشد", error_code="point_account_not_found"
|
|
)
|
|
else:
|
|
account = await self.accounts.get_by_member(tenant_id, member.id)
|
|
if account is None:
|
|
raise NotFoundError(
|
|
"حساب امتیاز یافت نشد", error_code="point_account_not_found"
|
|
)
|
|
|
|
rules = campaign.rules or {}
|
|
max_applications = rules.get("max_applications_per_member")
|
|
existing_count = await self.applications.count_for_member(
|
|
tenant_id,
|
|
campaign.id,
|
|
member.id,
|
|
exclude_statuses=(CampaignApplicationStatus.REVERSED,),
|
|
)
|
|
ensure_max_applications_per_member(existing_count, max_applications)
|
|
|
|
grant = ensure_positive_grant(compute_grant_amount(rules, body.base_amount))
|
|
|
|
actor_id = actor.user_id if actor else None
|
|
entry = await self.point_engine.earn(
|
|
tenant_id,
|
|
account.id,
|
|
PointEarnRequest(
|
|
amount=grant,
|
|
reason=f"campaign:{campaign.code}",
|
|
reference_type="campaign_application",
|
|
reference_id=str(campaign.id),
|
|
),
|
|
actor=actor,
|
|
auto_commit=False,
|
|
)
|
|
|
|
application = CampaignApplication(
|
|
tenant_id=tenant_id,
|
|
campaign_id=campaign.id,
|
|
program_id=campaign.program_id,
|
|
member_id=member.id,
|
|
point_account_id=account.id,
|
|
rule_version=campaign.rule_version,
|
|
points_granted=grant,
|
|
ledger_entry_id=entry.id,
|
|
status=CampaignApplicationStatus.APPLIED,
|
|
idempotency_key=body.idempotency_key,
|
|
actor_user_id=actor_id,
|
|
metadata_json=body.metadata_json,
|
|
)
|
|
try:
|
|
await self.applications.add(application)
|
|
except IntegrityError:
|
|
await self.session.rollback()
|
|
raise AppError(
|
|
"اعمال کمپین تکراری است",
|
|
status_code=409,
|
|
error_code="campaign_application_idempotency_conflict",
|
|
)
|
|
|
|
await self.audit.record(
|
|
tenant_id=tenant_id,
|
|
entity_type="campaign_application",
|
|
entity_id=application.id,
|
|
action=AuditAction.APPLY,
|
|
actor_user_id=actor_id,
|
|
message=(
|
|
f"Campaign {campaign.code} applied to member "
|
|
f"{member.membership_number} (+{grant})"
|
|
),
|
|
changes={"points_granted": grant, "campaign_id": str(campaign.id)},
|
|
)
|
|
await self.publisher.publish(
|
|
event_type=LoyaltyEventType.CAMPAIGN_APPLIED,
|
|
aggregate_type="campaign_application",
|
|
aggregate_id=application.id,
|
|
tenant_id=tenant_id,
|
|
payload={
|
|
"campaign_id": str(campaign.id),
|
|
"member_id": str(member.id),
|
|
"points_granted": grant,
|
|
},
|
|
)
|
|
await self.session.commit()
|
|
await self.session.refresh(application)
|
|
return application
|
|
|
|
async def list_applications(
|
|
self,
|
|
tenant_id: UUID,
|
|
campaign_id: UUID,
|
|
*,
|
|
offset: int = 0,
|
|
limit: int = 50,
|
|
):
|
|
await self.get(tenant_id, campaign_id)
|
|
rows = await self.applications.list_for_campaign(
|
|
tenant_id, campaign_id, offset=offset, limit=limit
|
|
)
|
|
total = await self.applications.count_for_campaign(tenant_id, campaign_id)
|
|
return rows, total
|