feat(domains): identity, billing, task services and domain placeholders
Billing implements reserve -> commit -> refund against bl_quota and bl_ledger with row-level locking and idempotency keys; the refund commits on its own transaction so a failed request cannot silently keep a user's quota. Task implements the unified task centre over the state machine. Identity issues and resolves users. Remaining domains are placeholders whose READMEs state what belongs there and what must not.
This commit is contained in:
@@ -0,0 +1,38 @@
|
||||
# `packages/billing/` — 配额与计量（V1）
|
||||
|
||||
> ⚠️ **V1 不收费**（TD-28）。本域在 V1 的职责是**配额限制 + 计量**，
|
||||
> 不含任何支付、订单、购买、会员能力。
|
||||
|
||||
## 已实现
|
||||
|
||||
- `bl_quota_rule`（规则配置化，改配置不发版，FR-BILL-04）
|
||||
- `bl_quota`（按 `period_key` 自然重置，无需定时任务清零）
|
||||
- `bl_ledger`（流水是真相，余额是缓存）
|
||||
- **预扣三段式** `reserve → commit → refund`（架构 §7.4，FR-BILL-05）
|
||||
- 服务端硬校验（FR-BILL-06）；退还幂等（refund 键由 reserve 键派生）
|
||||
|
||||
## 架构约束（不可妥协）
|
||||
|
||||
**V1 的配额账本与 V2 的点数账本必须是同一张表。** V1 只用 `system_grant` /
|
||||
`consume` / `refund` 三种 `tx_type`，V2 增加 `purchase` / `gift`。这样 V2 上
|
||||
支付时不需要数据迁移，也不需要重写扣减逻辑。**不要为了"V1 更简单"把表拆开。**
|
||||
|
||||
## 为什么是预扣不是直接扣
|
||||
|
||||
长耗时任务可能失败，直接扣会产生「扣了额度没拿到结果」的客诉。V1 虽不涉及
|
||||
金钱，**用户对"次数被白扣"的敏感度不比扣钱低**，而且 V1 没有客服退款兜底手段，
|
||||
只能靠机制保证。
|
||||
|
||||
## 待实现
|
||||
|
||||
- FR-BILL-03 任务奖励发放（签到/完善资料/反馈/邀请/分享，同任务同周期幂等）
|
||||
- **每日对账**：余额由流水重算校验，不一致则告警。V1 就要做，否则 V2 上支付时
|
||||
对账逻辑得从零开发且无历史数据验证。
|
||||
- 设备/账号/IP 三重限流（FR-BILL-06）
|
||||
- Redis 配额计数器 + PG 落账的两级结构（架构 §10）
|
||||
|
||||
## 文案约束（FR-BILL-07）
|
||||
|
||||
额度相关提示里**不得出现任何价格、购买、会员、升级字样**，只提供「等待恢复」
|
||||
与「做任务换额度」两条路径。`tests/unit/test_response_envelope.py` 会扫描全部
|
||||
错误码文案。
|
||||
@@ -0,0 +1 @@
|
||||
"""billing 域（V1）：配额账本、流水、预扣三段式。不含支付。"""
|
||||
Executable
+79
@@ -0,0 +1,79 @@
|
||||
"""Data access for the billing domain. **Only this module may touch bl_* tables.**
|
||||
|
||||
架构文档 §3.2 硬性规则 1: 每个域只能通过对方暴露的 service 接口调用，禁止直接
|
||||
查对方的表。import-linter enforces it (contract "no cross-domain repository access").
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from packages.shared.models import Ledger, Quota, QuotaRule
|
||||
|
||||
|
||||
async def get_rule(session: AsyncSession, quota_key: str) -> QuotaRule | None:
|
||||
return await session.get(QuotaRule, quota_key)
|
||||
|
||||
|
||||
async def lock_or_create_quota(
|
||||
session: AsyncSession,
|
||||
*,
|
||||
user_id: int,
|
||||
quota_key: str,
|
||||
period_key: str,
|
||||
limit_value: int,
|
||||
) -> Quota:
|
||||
"""Get the period row with a row lock, creating it on first use.
|
||||
|
||||
``ON CONFLICT DO NOTHING`` + ``SELECT ... FOR UPDATE`` makes concurrent first
|
||||
requests for the same period safe without an advisory lock.
|
||||
"""
|
||||
await session.execute(
|
||||
pg_insert(Quota)
|
||||
.values(
|
||||
user_id=user_id,
|
||||
quota_key=quota_key,
|
||||
period_key=period_key,
|
||||
used=0,
|
||||
limit_value=limit_value,
|
||||
bonus_value=0,
|
||||
)
|
||||
.on_conflict_do_nothing(index_elements=["user_id", "quota_key", "period_key"])
|
||||
)
|
||||
result = await session.execute(
|
||||
select(Quota)
|
||||
.where(
|
||||
Quota.user_id == user_id,
|
||||
Quota.quota_key == quota_key,
|
||||
Quota.period_key == period_key,
|
||||
)
|
||||
.with_for_update()
|
||||
)
|
||||
row = result.scalar_one()
|
||||
return row
|
||||
|
||||
|
||||
async def get_ledger_by_idempotency_key(session: AsyncSession, key: str) -> Ledger | None:
|
||||
result = await session.execute(select(Ledger).where(Ledger.idempotency_key == key))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
async def get_ledger(session: AsyncSession, tx_id: int) -> Ledger | None:
|
||||
return await session.get(Ledger, tx_id)
|
||||
|
||||
|
||||
async def list_ledger(session: AsyncSession, user_id: int, limit: int = 50) -> list[Ledger]:
|
||||
result = await session.execute(
|
||||
select(Ledger)
|
||||
.where(Ledger.user_id == user_id)
|
||||
.order_by(Ledger.created_at.desc(), Ledger.tx_id.desc())
|
||||
.limit(limit)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
|
||||
def add_ledger(session: AsyncSession, ledger: Ledger) -> Ledger:
|
||||
session.add(ledger)
|
||||
return ledger
|
||||
Executable
+267
@@ -0,0 +1,267 @@
|
||||
"""Quota three-phase protocol: reserve → commit → refund (架构文档 §7.4).
|
||||
|
||||
为什么预扣而不是直接扣: 长耗时任务可能失败，直接扣会产生「扣了额度没拿到
|
||||
结果」的客诉。V1 没有客服退款兜底手段，只能靠机制保证 (FR-BILL-05).
|
||||
|
||||
Nothing here decides *how much* a feature costs — that comes from
|
||||
``bl_quota_rule`` so it can change without a release (FR-BILL-04).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime
|
||||
from typing import cast
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from packages.billing import repo
|
||||
from packages.domain.billing import quota as quota_rules
|
||||
from packages.domain.errors import QuotaExceededError
|
||||
from packages.domain.ids import next_id
|
||||
from packages.shared.errors import AppError, ErrorCode, QuotaExhaustedError
|
||||
from packages.shared.logging import get_logger
|
||||
from packages.shared.models import Ledger, QuotaRule
|
||||
from packages.shared.observability import QUOTA_EVENTS
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Reservation:
|
||||
tx_id: int
|
||||
user_id: int
|
||||
quota_key: str
|
||||
period_key: str
|
||||
amount: int
|
||||
remaining: int
|
||||
idempotency_key: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class QuotaSnapshot:
|
||||
quota_key: str
|
||||
period_key: str
|
||||
used: int
|
||||
limit_value: int
|
||||
bonus_value: int
|
||||
remaining: int
|
||||
reset_at: datetime | None
|
||||
|
||||
|
||||
class BillingService:
|
||||
def __init__(self, session: AsyncSession, *, tz_offset_hours: int = 8) -> None:
|
||||
self._session = session
|
||||
self._tz_offset = tz_offset_hours
|
||||
|
||||
# -- read ---------------------------------------------------------------
|
||||
|
||||
async def get_quota(self, *, user_id: int, quota_key: str) -> QuotaSnapshot:
|
||||
rule = await self._require_rule(quota_key)
|
||||
period_type = cast(quota_rules.PeriodType, rule.period_type)
|
||||
now = datetime.now(UTC)
|
||||
period_key = quota_rules.period_key(now, period_type, self._tz_offset)
|
||||
row = await repo.lock_or_create_quota(
|
||||
self._session,
|
||||
user_id=user_id,
|
||||
quota_key=quota_key,
|
||||
period_key=period_key,
|
||||
limit_value=rule.base_limit,
|
||||
)
|
||||
state = quota_rules.QuotaState(
|
||||
used=row.used, base_limit=row.limit_value, bonus=row.bonus_value
|
||||
)
|
||||
return QuotaSnapshot(
|
||||
quota_key=quota_key,
|
||||
period_key=period_key,
|
||||
used=row.used,
|
||||
limit_value=row.limit_value,
|
||||
bonus_value=row.bonus_value,
|
||||
remaining=state.remaining,
|
||||
reset_at=quota_rules.period_reset_at(now, period_type, self._tz_offset),
|
||||
)
|
||||
|
||||
async def list_ledger(self, *, user_id: int, limit: int = 50) -> list[Ledger]:
|
||||
return await repo.list_ledger(self._session, user_id, limit)
|
||||
|
||||
# -- phase 1: reserve ---------------------------------------------------
|
||||
|
||||
async def reserve(
|
||||
self,
|
||||
*,
|
||||
user_id: int,
|
||||
quota_key: str,
|
||||
request_key: str,
|
||||
amount: int = 1,
|
||||
reason: str | None = None,
|
||||
ref_type: str | None = None,
|
||||
ref_id: str | None = None,
|
||||
) -> Reservation:
|
||||
"""Freeze ``amount`` units. Idempotent on ``request_key``."""
|
||||
idem_key = quota_rules.reserve_idempotency_key(
|
||||
user_id=user_id, quota_key=quota_key, request_key=request_key
|
||||
)
|
||||
existing = await repo.get_ledger_by_idempotency_key(self._session, idem_key)
|
||||
if existing is not None:
|
||||
QUOTA_EVENTS.labels(quota_key, "reserve", "idempotent_hit").inc()
|
||||
return Reservation(
|
||||
tx_id=existing.tx_id,
|
||||
user_id=user_id,
|
||||
quota_key=quota_key,
|
||||
period_key=existing.period_key or "",
|
||||
amount=abs(existing.delta),
|
||||
remaining=existing.balance_after,
|
||||
idempotency_key=idem_key,
|
||||
)
|
||||
|
||||
rule = await self._require_rule(quota_key)
|
||||
period_type = cast(quota_rules.PeriodType, rule.period_type)
|
||||
now = datetime.now(UTC)
|
||||
period_key = quota_rules.period_key(now, period_type, self._tz_offset)
|
||||
row = await repo.lock_or_create_quota(
|
||||
self._session,
|
||||
user_id=user_id,
|
||||
quota_key=quota_key,
|
||||
period_key=period_key,
|
||||
limit_value=rule.base_limit,
|
||||
)
|
||||
state = quota_rules.QuotaState(
|
||||
used=row.used, base_limit=row.limit_value, bonus=row.bonus_value
|
||||
)
|
||||
|
||||
try:
|
||||
new_state = quota_rules.reserve(state, amount)
|
||||
except QuotaExceededError as exc:
|
||||
QUOTA_EVENTS.labels(quota_key, "reserve", "exhausted").inc()
|
||||
reset_at = quota_rules.period_reset_at(now, period_type, self._tz_offset)
|
||||
raise QuotaExhaustedError(
|
||||
required=exc.required,
|
||||
remaining=exc.remaining,
|
||||
reset_at=reset_at.isoformat() if reset_at else None,
|
||||
) from exc
|
||||
|
||||
row.used = new_state.used
|
||||
tx_id = next_id()
|
||||
repo.add_ledger(
|
||||
self._session,
|
||||
Ledger(
|
||||
tx_id=tx_id,
|
||||
user_id=user_id,
|
||||
delta=-amount,
|
||||
tx_type="consume",
|
||||
status="reserved",
|
||||
reason=reason,
|
||||
ref_type=ref_type,
|
||||
ref_id=ref_id,
|
||||
quota_key=quota_key,
|
||||
period_key=period_key,
|
||||
balance_after=quota_rules.balance_after(new_state),
|
||||
idempotency_key=idem_key,
|
||||
),
|
||||
)
|
||||
await self._session.flush()
|
||||
QUOTA_EVENTS.labels(quota_key, "reserve", "ok").inc()
|
||||
logger.info(
|
||||
"quota_reserved",
|
||||
tx_id=tx_id,
|
||||
quota_key=quota_key,
|
||||
amount=amount,
|
||||
remaining=new_state.remaining,
|
||||
)
|
||||
return Reservation(
|
||||
tx_id=tx_id,
|
||||
user_id=user_id,
|
||||
quota_key=quota_key,
|
||||
period_key=period_key,
|
||||
amount=amount,
|
||||
remaining=new_state.remaining,
|
||||
idempotency_key=idem_key,
|
||||
)
|
||||
|
||||
# -- phase 2: commit ----------------------------------------------------
|
||||
|
||||
async def commit(self, reservation: Reservation) -> None:
|
||||
"""Confirm the deduction. Safe to call twice."""
|
||||
ledger = await repo.get_ledger(self._session, reservation.tx_id)
|
||||
if ledger is None:
|
||||
raise AppError(ErrorCode.STATE_CONFLICT, message="配额流水不存在")
|
||||
if ledger.status == "reserved":
|
||||
ledger.status = "committed"
|
||||
await self._session.flush()
|
||||
QUOTA_EVENTS.labels(reservation.quota_key, "commit", "ok").inc()
|
||||
logger.info("quota_committed", tx_id=ledger.tx_id)
|
||||
else:
|
||||
QUOTA_EVENTS.labels(reservation.quota_key, "commit", "noop").inc()
|
||||
|
||||
# -- phase 3: refund ----------------------------------------------------
|
||||
|
||||
async def refund(self, reservation: Reservation, *, reason: str = "task_failed") -> None:
|
||||
"""Release a reservation and write the reverse ledger entry.
|
||||
|
||||
FR-BILL-05 requires refunds to be idempotent; the refund ledger key is
|
||||
derived from the reserve key so a retry can never double-refund.
|
||||
"""
|
||||
ledger = await repo.get_ledger(self._session, reservation.tx_id)
|
||||
if ledger is None:
|
||||
raise AppError(ErrorCode.STATE_CONFLICT, message="配额流水不存在")
|
||||
if ledger.status != "reserved":
|
||||
QUOTA_EVENTS.labels(reservation.quota_key, "refund", "noop").inc()
|
||||
return
|
||||
|
||||
refund_key = quota_rules.refund_idempotency_key(reservation.idempotency_key)
|
||||
if await repo.get_ledger_by_idempotency_key(self._session, refund_key) is not None:
|
||||
QUOTA_EVENTS.labels(reservation.quota_key, "refund", "noop").inc()
|
||||
return
|
||||
|
||||
rule = await self._require_rule(reservation.quota_key)
|
||||
row = await repo.lock_or_create_quota(
|
||||
self._session,
|
||||
user_id=reservation.user_id,
|
||||
quota_key=reservation.quota_key,
|
||||
period_key=reservation.period_key,
|
||||
limit_value=rule.base_limit,
|
||||
)
|
||||
state = quota_rules.QuotaState(
|
||||
used=row.used, base_limit=row.limit_value, bonus=row.bonus_value
|
||||
)
|
||||
new_state = quota_rules.refund(state, reservation.amount)
|
||||
row.used = new_state.used
|
||||
ledger.status = "refunded"
|
||||
|
||||
repo.add_ledger(
|
||||
self._session,
|
||||
Ledger(
|
||||
tx_id=next_id(),
|
||||
user_id=reservation.user_id,
|
||||
delta=reservation.amount,
|
||||
tx_type="refund",
|
||||
status="committed",
|
||||
reason=reason,
|
||||
ref_type=ledger.ref_type,
|
||||
ref_id=ledger.ref_id,
|
||||
quota_key=reservation.quota_key,
|
||||
period_key=reservation.period_key,
|
||||
balance_after=quota_rules.balance_after(new_state),
|
||||
idempotency_key=refund_key,
|
||||
),
|
||||
)
|
||||
await self._session.flush()
|
||||
QUOTA_EVENTS.labels(reservation.quota_key, "refund", "ok").inc()
|
||||
logger.info(
|
||||
"quota_refunded",
|
||||
tx_id=ledger.tx_id,
|
||||
amount=reservation.amount,
|
||||
remaining=new_state.remaining,
|
||||
)
|
||||
|
||||
# -- helpers ------------------------------------------------------------
|
||||
|
||||
async def _require_rule(self, quota_key: str) -> QuotaRule:
|
||||
rule = await repo.get_rule(self._session, quota_key)
|
||||
if rule is None:
|
||||
raise AppError(
|
||||
ErrorCode.CONFIG_ERROR,
|
||||
message="配额规则未配置",
|
||||
data={"quota_key": quota_key},
|
||||
)
|
||||
return rule
|
||||
Reference in New Issue
Block a user