public paste
Backend
—1 views131,498 chars
edit key captured
UPBqXRh1q// shown only once — save it securely
stdout
Paste content
# backend_v2.py
from __future__ import annotations
import asyncio
import hashlib
import hmac
import json
import logging
import os
import secrets
import signal
import time
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone, timedelta
from typing import Optional, Any
import httpx
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from fastapi import (
FastAPI,
Request,
Depends,
HTTPException,
Header,
)
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from redis.asyncio import Redis
from sqlalchemy import (
Boolean,
DateTime,
ForeignKey,
Integer,
String,
Text,
BigInteger,
UniqueConstraint,
Index,
select,
func,
update,
)
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.orm import (
DeclarativeBase,
Mapped,
mapped_column,
)
# ============================================================
# SETTINGS
# ============================================================
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
extra="ignore",
)
APP_NAME: str = "Follower Bگیر"
ENVIRONMENT: str = "production"
HOST: str = "0.0.0.0"
PORT: int = 8000
DATABASE_URL: str = (
"postgresql+asyncpg://postgres:[email protected]:5432/follower"
)
REDIS_URL: str = "redis://127.0.0.1:6379/0"
MASTER_KEY: str = ""
ADMIN_TOKEN: str = ""
CORS_ORIGINS: str = (
"http://127.0.0.1:8080,"
"http://localhost:8080"
)
API_VERSION: str = "1"
TOKEN_TTL: int = 30 * 24 * 3600
LOGIN_TTL: int = 600
MAX_LOGIN_ATTEMPTS: int = 5
REQUEST_WINDOW: int = 120
MIN_ORDER: int = 50
FOLLOW_REWARD: int = 1
PRIMARY_FOLLOW_BONUS: int = 2
UNFOLLOW_PENALTY: int = 2
DAILY_REWARD: int = 10
TRANSFER_MIN: int = 10
TRANSFER_TAX_PERCENT: int = 10
LUCKY_EVERY: int = 50
REQUEST_TIMEOUT: float = 20.0
MAX_BODY_SIZE: int = 1_000_000
settings = Settings()
if not settings.MASTER_KEY:
settings.MASTER_KEY = secrets.token_urlsafe(32)
# ============================================================
# LOGGING
# ============================================================
logging.basicConfig(
level=logging.INFO,
format=(
"%(asctime)s "
"%(levelname)s "
"%(name)s "
"%(message)s"
),
)
logger = logging.getLogger("follower_backend")
# ============================================================
# TIME
# ============================================================
def now_utc() -> datetime:
return datetime.now(timezone.utc)
def unix_time() -> int:
return int(time.time())
# ============================================================
# SECURITY
# ============================================================
def hash_token(token: str) -> str:
return hashlib.sha256(
token.encode("utf-8")
).hexdigest()
def hash_device(
ip: str,
device_id: str,
) -> str:
raw = f"{ip}|{device_id}"
return hashlib.sha256(
raw.encode("utf-8")
).hexdigest()
def encrypt_secret(value: str) -> str:
key = hashlib.sha256(
settings.MASTER_KEY.encode()
).digest()
nonce = secrets.token_bytes(12)
cipher = AESGCM(key)
encrypted = cipher.encrypt(
nonce,
value.encode(),
b"FOLLOWER-BACKEND-V2",
)
return (nonce + encrypted).hex()
def decrypt_secret(value: str) -> str:
raw = bytes.fromhex(value)
key = hashlib.sha256(
settings.MASTER_KEY.encode()
).digest()
cipher = AESGCM(key)
return cipher.decrypt(
raw[:12],
raw[12:],
b"FOLLOWER-BACKEND-V2",
).decode()
def validate_timestamp(
timestamp: int,
window: int | None = None,
):
if window is None:
window = settings.REQUEST_WINDOW
if abs(unix_time() - timestamp) > window:
raise HTTPException(
status_code=400,
detail="request_expired",
)
# ============================================================
# DATABASE
# ============================================================
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
username: Mapped[Optional[str]] = mapped_column(
String(100)
)
coins: Mapped[int] = mapped_column(
BigInteger,
default=0,
nullable=False,
)
lifetime_coins_earned: Mapped[int] = mapped_column(
BigInteger,
default=0,
nullable=False,
)
debt: Mapped[int] = mapped_column(
BigInteger,
default=0,
nullable=False,
)
is_banned: Mapped[bool] = mapped_column(
Boolean,
default=False,
nullable=False,
)
is_admin: Mapped[bool] = mapped_column(
Boolean,
default=False,
nullable=False,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
last_activity: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class Account(Base):
__tablename__ = "accounts"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
user_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
),
index=True,
)
phone: Mapped[str] = mapped_column(
String(32),
index=True,
)
auth_enc: Mapped[Optional[str]] = mapped_column(
Text
)
session_enc: Mapped[Optional[str]] = mapped_column(
Text
)
api_url: Mapped[Optional[str]] = mapped_column(
Text
)
device_hash: Mapped[Optional[str]] = mapped_column(
String(128)
)
is_active: Mapped[bool] = mapped_column(
Boolean,
default=True,
)
is_primary: Mapped[bool] = mapped_column(
Boolean,
default=False,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
__table_args__ = (
UniqueConstraint(
"user_id",
"phone",
name="uq_account_user_phone",
),
)
class LoginState(Base):
__tablename__ = "login_states"
login_id: Mapped[str] = mapped_column(
String(64),
primary_key=True,
)
user_id: Mapped[Optional[int]] = mapped_column(
ForeignKey(
"users.id",
ondelete="SET NULL",
)
)
phone: Mapped[str] = mapped_column(
String(32)
)
tmp_auth_enc: Mapped[Optional[str]] = mapped_column(
Text
)
api_url: Mapped[Optional[str]] = mapped_column(
Text
)
phone_code_hash: Mapped[Optional[str]] = mapped_column(
Text
)
status: Mapped[str] = mapped_column(
String(32),
default="code",
)
attempts_2fa: Mapped[int] = mapped_column(
Integer,
default=0,
)
attempts_code: Mapped[int] = mapped_column(
Integer,
default=0,
)
request_id: Mapped[str] = mapped_column(
String(128)
)
device_hash: Mapped[str] = mapped_column(
String(128)
)
ip: Mapped[str] = mapped_column(
String(64)
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
expires_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True)
)
class ApiToken(Base):
__tablename__ = "api_tokens"
token_hash: Mapped[str] = mapped_column(
String(128),
primary_key=True,
)
user_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
),
index=True,
)
device_hash: Mapped[str] = mapped_column(
String(128)
)
expires_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True)
)
revoked: Mapped[bool] = mapped_column(
Boolean,
default=False,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class Order(Base):
__tablename__ = "orders"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
user_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
),
index=True,
)
account_id: Mapped[int] = mapped_column(
ForeignKey(
"accounts.id",
ondelete="RESTRICT",
)
)
target_username: Mapped[str] = mapped_column(
String(128)
)
target_profile_id: Mapped[str] = mapped_column(
String(128)
)
quantity: Mapped[int] = mapped_column(
Integer
)
successful_count: Mapped[int] = mapped_column(
Integer,
default=0,
)
status: Mapped[str] = mapped_column(
String(32),
default="pending",
index=True,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class OrderAction(Base):
__tablename__ = "order_actions"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
order_id: Mapped[int] = mapped_column(
ForeignKey(
"orders.id",
ondelete="CASCADE",
),
index=True,
)
user_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
)
)
account_id: Mapped[int] = mapped_column(
ForeignKey(
"accounts.id",
ondelete="CASCADE",
)
)
follower_profile_id: Mapped[str] = mapped_column(
String(128)
)
status: Mapped[str] = mapped_column(
String(32),
default="processing",
)
reward: Mapped[int] = mapped_column(
Integer,
default=0,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
__table_args__ = (
UniqueConstraint(
"order_id",
"user_id",
"follower_profile_id",
name="uq_order_follower",
),
)
class CoinTransaction(Base):
__tablename__ = "coin_transactions"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
user_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
),
index=True,
)
account_id: Mapped[Optional[int]] = mapped_column(
ForeignKey(
"accounts.id",
ondelete="SET NULL",
)
)
amount: Mapped[int] = mapped_column(
BigInteger
)
balance_after: Mapped[int] = mapped_column(
BigInteger
)
reason: Mapped[str] = mapped_column(
String(128)
)
reference_id: Mapped[Optional[str]] = mapped_column(
String(128)
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class UsedRequest(Base):
__tablename__ = "used_requests"
request_id: Mapped[str] = mapped_column(
String(128),
primary_key=True,
)
user_id: Mapped[Optional[int]] = mapped_column(
ForeignKey(
"users.id",
ondelete="SET NULL",
)
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class Notification(Base):
__tablename__ = "notifications"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
user_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
),
index=True,
)
title: Mapped[str] = mapped_column(
String(200)
)
body: Mapped[str] = mapped_column(
Text
)
is_read: Mapped[bool] = mapped_column(
Boolean,
default=False,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class SupportTicket(Base):
__tablename__ = "support_tickets"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
user_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
)
)
subject: Mapped[str] = mapped_column(
String(200)
)
status: Mapped[str] = mapped_column(
String(32),
default="open",
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class SupportMessage(Base):
__tablename__ = "support_messages"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
ticket_id: Mapped[int] = mapped_column(
ForeignKey(
"support_tickets.id",
ondelete="CASCADE",
)
)
user_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
)
)
message: Mapped[str] = mapped_column(
Text
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class ReferralUse(Base):
__tablename__ = "referral_uses"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
referrer_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
)
)
referred_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
),
unique=True,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class Unfollow(Base):
__tablename__ = "unfollows"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
order_action_id: Mapped[int] = mapped_column(
ForeignKey(
"order_actions.id",
ondelete="CASCADE",
)
)
user_id: Mapped[int] = mapped_column(
ForeignKey(
"users.id",
ondelete="CASCADE",
)
)
penalty: Mapped[int] = mapped_column(
Integer
)
detected_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class Job(Base):
__tablename__ = "jobs"
id: Mapped[str] = mapped_column(
String(64),
primary_key=True,
)
kind: Mapped[str] = mapped_column(
String(64),
index=True,
)
payload: Mapped[str] = mapped_column(
Text
)
status: Mapped[str] = mapped_column(
String(32),
default="pending",
index=True,
)
attempts: Mapped[int] = mapped_column(
Integer,
default=0,
)
available_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
error: Mapped[Optional[str]] = mapped_column(
Text
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class AuditLog(Base):
__tablename__ = "audit_logs"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
user_id: Mapped[Optional[int]] = mapped_column(
ForeignKey(
"users.id",
ondelete="SET NULL",
)
)
action: Mapped[str] = mapped_column(
String(128)
)
ip: Mapped[Optional[str]] = mapped_column(
String(64)
)
correlation_id: Mapped[str] = mapped_column(
String(64)
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
class SecurityEvent(Base):
__tablename__ = "security_events"
id: Mapped[int] = mapped_column(
BigInteger,
primary_key=True,
)
event_type: Mapped[str] = mapped_column(
String(128)
)
severity: Mapped[str] = mapped_column(
String(32)
)
ip: Mapped[Optional[str]] = mapped_column(
String(64)
)
device_hash: Mapped[Optional[str]] = mapped_column(
String(128)
)
details: Mapped[Optional[str]] = mapped_column(
Text
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
# ============================================================
# DATABASE CONNECTION
# ============================================================
engine = create_async_engine(
settings.DATABASE_URL,
pool_pre_ping=True,
pool_size=5,
max_overflow=5,
pool_recycle=1800,
)
SessionLocal = async_sessionmaker(
engine,
expire_on_commit=False,
)
# ============================================================
# REDIS
# ============================================================
redis = Redis.from_url(
settings.REDIS_URL,
decode_responses=True,
)
# ============================================================
# PYDANTIC MODELS
# ============================================================
class StartLogin(BaseModel):
phone: str = Field(
min_length=8,
max_length=32,
)
request_id: str = Field(
min_length=8,
max_length=128,
)
timestamp: int
device_id: str = Field(
min_length=8,
max_length=256,
)
class Verify2FA(BaseModel):
login_id: str
phone: str
password: str = Field(
min_length=1,
max_length=256,
)
request_id: str
timestamp: int
device_id: str
class VerifyCode(BaseModel):
login_id: str
phone: str
code: str = Field(
min_length=3,
max_length=32,
)
request_id: str
timestamp: int
device_id: str
referral_code: Optional[str] = None
class CreateOrder(BaseModel):
account_id: int
target_username: str = Field(
min_length=1,
max_length=128,
)
target_profile_id: str = Field(
min_length=1,
max_length=128,
)
quantity: int = Field(
ge=settings.MIN_ORDER,
le=1_000_000,
)
class FollowRequest(BaseModel):
follower_profile_id: str = Field(
min_length=1,
max_length=128,
)
request_id: str
timestamp: int
class TransferRequest(BaseModel):
receiver_account_id: int
amount: int = Field(
ge=settings.TRANSFER_MIN
)
request_id: str
timestamp: int
class ClaimRequest(BaseModel):
request_id: str
timestamp: int
class TicketRequest(BaseModel):
subject: str = Field(
min_length=1,
max_length=200,
)
message: str = Field(
min_length=1,
max_length=5000,
)
request_id: str
timestamp: int
class MessageRequest(BaseModel):
message: str = Field(
min_length=1,
max_length=5000,
)
request_id: str
timestamp: int
class ReferralRequest(BaseModel):
code: str = Field(
min_length=1,
max_length=64,
)
request_id: str
timestamp: int
# ============================================================
# SECTION 2
# APP / MIDDLEWARE / AUTH / ACCOUNTS / COINS
# PROFILES / ORDERS / FOLLOW / REWARDS / NOTIFICATIONS
# ============================================================
# ============================================================
# FASTAPI APP / LIFESPAN
# ============================================================
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
@asynccontextmanager
async def lifespan(app: FastAPI):
await init_db()
try:
await redis.ping()
logger.info("redis_connected")
except Exception as exc:
logger.warning("redis_unavailable: %s", exc)
logger.info("backend_started")
yield
try:
await redis.close()
except Exception:
pass
await engine.dispose()
logger.info("backend_stopped")
app = FastAPI(
title=settings.APP_NAME,
version=settings.API_VERSION,
lifespan=lifespan,
)
# ============================================================
# CORS
# ============================================================
app.add_middleware(
CORSMiddleware,
allow_origins=[
x.strip()
for x in settings.CORS_ORIGINS.split(",")
if x.strip()
],
allow_credentials=True,
allow_methods=[
"GET",
"POST",
"PUT",
"PATCH",
"DELETE",
"OPTIONS",
],
allow_headers=[
"Authorization",
"Content-Type",
"X-Request-ID",
"X-Device-ID",
"X-Client-Version",
"Idempotency-Key",
],
)
# ============================================================
# SECURITY / REQUEST MIDDLEWARE
# ============================================================
@app.middleware("http")
async def security_middleware(
request: Request,
call_next,
):
request_id = (
request.headers.get("X-Request-ID")
or str(uuid.uuid4())
)
request.state.request_id = request_id
# Request body size
content_length = request.headers.get("Content-Length")
if content_length:
try:
body_size = int(content_length)
if body_size > settings.MAX_BODY_SIZE:
return JSONResponse(
status_code=413,
content={
"ok": False,
"error": "request_too_large",
"request_id": request_id,
},
)
except ValueError:
pass
started = time.perf_counter()
try:
response = await call_next(request)
except HTTPException:
raise
except Exception:
logger.exception(
"unhandled_request_error "
"request_id=%s path=%s",
request_id,
request.url.path,
)
return JSONResponse(
status_code=500,
content={
"ok": False,
"error": "internal_server_error",
"request_id": request_id,
},
)
elapsed = time.perf_counter() - started
if elapsed >= 2:
logger.warning(
"slow_request "
"request_id=%s path=%s elapsed=%.3f",
request_id,
request.url.path,
elapsed,
)
response.headers["X-Request-ID"] = request_id
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["Referrer-Policy"] = "no-referrer"
response.headers["Cache-Control"] = "no-store"
return response
# ============================================================
# RATE LIMIT
# ============================================================
async def rate_limit(
key: str,
limit: int,
window: int,
):
bucket = int(time.time()) // window
redis_key = f"rl:{key}:{bucket}"
try:
count = await redis.incr(redis_key)
if count == 1:
await redis.expire(
redis_key,
window + 2,
)
if count > limit:
raise HTTPException(
status_code=429,
detail="rate_limit_exceeded",
)
except HTTPException:
raise
except Exception:
# اگر Redis موقتاً در دسترس نبود،
# درخواست را کاملاً متوقف نمیکنیم.
logger.warning(
"rate_limit_redis_unavailable"
)
# ============================================================
# REQUEST ID REPLAY PROTECTION
# ============================================================
async def consume_request_id(
request_id: str,
user_id: Optional[int] = None,
):
if not request_id:
raise HTTPException(
status_code=400,
detail="request_id_required",
)
key = f"request-id:{request_id}"
try:
created = await redis.set(
key,
str(user_id or 0),
nx=True,
ex=settings.REQUEST_WINDOW,
)
if not created:
raise HTTPException(
status_code=409,
detail="duplicate_request",
)
except HTTPException:
raise
except Exception:
logger.warning(
"request_id_redis_unavailable"
)
# ============================================================
# CLIENT IP
# ============================================================
def get_client_ip(request: Request) -> str:
forwarded = request.headers.get(
"X-Forwarded-For"
)
if forwarded:
return forwarded.split(",")[0].strip()
return request.client.host if request.client else "unknown"
# ============================================================
# DEVICE HASH
# ============================================================
def get_device_hash(
request: Request,
) -> str:
device_id = (
request.headers.get("X-Device-ID")
or "unknown-device"
)
ip = get_client_ip(request)
return hash_device(
f"{ip}|{device_id}"
)
# ============================================================
# CURRENT USER
# ============================================================
async def current_user(
request: Request,
authorization: Optional[str] = Header(
default=None
),
) -> User:
if not authorization:
raise HTTPException(
status_code=401,
detail="authentication_required",
)
if not authorization.lower().startswith(
"bearer "
):
raise HTTPException(
status_code=401,
detail="invalid_authorization",
)
raw_token = authorization[7:].strip()
if not raw_token:
raise HTTPException(
status_code=401,
detail="invalid_token",
)
token_hash = hash_token(raw_token)
# revoked token
try:
revoked = await redis.get(
f"revoked-token:{token_hash}"
)
if revoked:
raise HTTPException(
status_code=401,
detail="token_revoked",
)
except HTTPException:
raise
except Exception:
logger.warning(
"redis_revocation_check_failed"
)
device_hash = get_device_hash(request)
async with SessionLocal() as db:
result = await db.execute(
select(User, ApiToken)
.join(
ApiToken,
ApiToken.user_id == User.id,
)
.where(
ApiToken.token_hash == token_hash
)
)
row = result.first()
if not row:
raise HTTPException(
status_code=401,
detail="invalid_token",
)
user, token = row
now = now_utc()
if token.expires_at <= now:
raise HTTPException(
status_code=401,
detail="token_expired",
)
if not token.is_active:
raise HTTPException(
status_code=401,
detail="token_inactive",
)
if user.is_banned:
raise HTTPException(
status_code=403,
detail="account_banned",
)
if token.device_hash:
if not hmac.compare_digest(
token.device_hash,
device_hash,
):
raise HTTPException(
status_code=401,
detail="device_mismatch",
)
token.last_activity = now
await db.commit()
return user
# ============================================================
# ADMIN
# ============================================================
async def require_admin(
request: Request,
):
token = request.headers.get(
"X-Admin-Token"
)
if not token:
raise HTTPException(
status_code=403,
detail="admin_required",
)
if not settings.ADMIN_TOKEN:
raise HTTPException(
status_code=503,
detail="admin_not_configured",
)
if not hmac.compare_digest(
token,
settings.ADMIN_TOKEN,
):
raise HTTPException(
status_code=403,
detail="invalid_admin_token",
)
return True
# ============================================================
# RUBIKA CLIENT
# ============================================================
class RubikaClient:
def __init__(self):
self.timeout = httpx.Timeout(
settings.REQUEST_TIMEOUT
)
async def get_api_server(self) -> str:
fallback = (
"https://getdcmess.iranlms.ir"
)
try:
async with httpx.AsyncClient(
timeout=self.timeout
) as client:
response = await client.get(
fallback
)
response.raise_for_status()
data = response.json()
if isinstance(data, str):
return data.rstrip("/")
if isinstance(data, dict):
for key in (
"api_url",
"api",
"server",
"url",
):
value = data.get(key)
if value:
return str(
value
).rstrip("/")
except Exception:
logger.warning(
"rubika_api_server_lookup_failed"
)
return fallback
async def start_login(
self,
phone: str,
):
raise RuntimeError(
"rubika_protocol_not_connected"
)
async def verify_2fa(
self,
auth: str,
password: str,
):
raise RuntimeError(
"rubika_protocol_not_connected"
)
async def verify_code(
self,
auth: str,
phone_code_hash: str,
code: str,
):
raise RuntimeError(
"rubika_protocol_not_connected"
)
async def profiles(
self,
account: Account,
):
raise RuntimeError(
"rubika_protocol_not_connected"
)
async def follow(
self,
account: Account,
profile_id: str,
):
raise RuntimeError(
"rubika_protocol_not_connected"
)
async def is_following(
self,
account: Account,
profile_id: str,
):
raise RuntimeError(
"rubika_protocol_not_connected"
)
rubika = RubikaClient()
# ============================================================
# COIN HELPER
# ============================================================
async def add_coins(
db: AsyncSession,
user_id: int,
amount: int,
transaction_type: str,
reason: str,
account_id: Optional[int] = None,
request_id: Optional[str] = None,
):
if amount == 0:
return
db.add(
CoinTransaction(
user_id=user_id,
account_id=account_id,
amount=amount,
type=transaction_type,
reason=reason,
request_id=request_id,
)
)
# ============================================================
# JOB HELPER
# ============================================================
async def enqueue_job(
db: AsyncSession,
job_type: str,
payload: dict,
priority: int = 100,
):
job = Job(
type=job_type,
payload=json.dumps(
payload,
ensure_ascii=False,
),
status="pending",
priority=priority,
)
db.add(job)
await db.flush()
return job
# ============================================================
# AUTH START
# ============================================================
@app.post("/api/v1/auth/start")
async def auth_start(
payload: StartLogin,
request: Request,
):
ip = get_client_ip(request)
device_hash = get_device_hash(request)
await rate_limit(
f"login-ip:{ip}",
10,
60,
)
await rate_limit(
f"login-phone:{payload.phone}",
5,
60,
)
login_id = secrets.token_urlsafe(32)
api_url = await rubika.get_api_server()
try:
result = await rubika.start_login(
payload.phone
)
except RuntimeError as exc:
if str(exc) == "rubika_protocol_not_connected":
raise HTTPException(
status_code=503,
detail="rubika_protocol_not_connected",
)
raise HTTPException(
status_code=502,
detail="rubika_login_failed",
)
except Exception:
logger.exception(
"auth_start_failed"
)
raise HTTPException(
status_code=502,
detail="rubika_login_failed",
)
async with SessionLocal() as db:
expires = now_utc() + timedelta(
seconds=settings.LOGIN_TTL
)
state = LoginState(
login_id=login_id,
user_id=None,
phone=payload.phone,
tmp_auth=encrypt_secret(
str(result.get("tmp_auth", ""))
),
tmp_key=encrypt_secret(
str(result.get("tmp_key", ""))
),
api_url=api_url,
phone_code_hash=encrypt_secret(
str(
result.get(
"phone_code_hash",
"",
)
)
),
status="code_sent",
attempts_2fa=0,
attempts_code=0,
created_at=now_utc(),
expires_at=expires,
request_id=request.headers.get(
"X-Request-ID"
) or str(uuid.uuid4()),
device_id=(
request.headers.get(
"X-Device-ID"
) or ""
),
device_hash=device_hash,
ip=ip,
)
db.add(state)
await db.commit()
return {
"ok": True,
"login_id": login_id,
"expires_in": settings.LOGIN_TTL,
"requires_2fa": bool(
result.get(
"requires_2fa",
False,
)
),
}
# ============================================================
# VERIFY 2FA
# ============================================================
@app.post("/api/v1/auth/verify-2fa")
async def verify_2fa(
payload: Verify2FA,
request: Request,
):
device_hash = get_device_hash(request)
async with SessionLocal() as db:
state = (
await db.execute(
select(LoginState).where(
LoginState.login_id
== payload.login_id
)
)
).scalar_one_or_none()
if not state:
raise HTTPException(
status_code=404,
detail="login_not_found",
)
if state.expires_at <= now_utc():
raise HTTPException(
status_code=400,
detail="login_expired",
)
if state.device_hash != device_hash:
raise HTTPException(
status_code=403,
detail="device_mismatch",
)
if state.attempts_2fa >= settings.MAX_LOGIN_ATTEMPTS:
raise HTTPException(
status_code=429,
detail="too_many_attempts",
)
state.attempts_2fa += 1
try:
auth = decrypt_secret(
state.tmp_auth
)
result = await rubika.verify_2fa(
auth,
payload.password,
)
except RuntimeError as exc:
if str(exc) == "rubika_protocol_not_connected":
raise HTTPException(
status_code=503,
detail="rubika_protocol_not_connected",
)
raise HTTPException(
status_code=400,
detail="invalid_2fa",
)
except Exception:
logger.exception(
"verify_2fa_failed"
)
raise HTTPException(
status_code=400,
detail="invalid_2fa",
)
state.tmp_auth = encrypt_secret(
str(
result.get(
"tmp_auth",
auth,
)
)
)
state.status = "code_sent"
await db.commit()
return {
"ok": True,
"login_id": payload.login_id,
"status": state.status,
}
# ============================================================
# VERIFY CODE
# ============================================================
@app.post("/api/v1/auth/verify-code")
async def verify_code(
payload: VerifyCode,
request: Request,
):
device_hash = get_device_hash(request)
async with SessionLocal() as db:
state = (
await db.execute(
select(LoginState).where(
LoginState.login_id
== payload.login_id
)
)
).scalar_one_or_none()
if not state:
raise HTTPException(
status_code=404,
detail="login_not_found",
)
if state.expires_at <= now_utc():
raise HTTPException(
status_code=400,
detail="login_expired",
)
if state.device_hash != device_hash:
raise HTTPException(
status_code=403,
detail="device_mismatch",
)
if state.attempts_code >= settings.MAX_LOGIN_ATTEMPTS:
raise HTTPException(
status_code=429,
detail="too_many_attempts",
)
state.attempts_code += 1
try:
auth = decrypt_secret(
state.tmp_auth
)
phone_code_hash = decrypt_secret(
state.phone_code_hash
)
result = await rubika.verify_code(
auth,
phone_code_hash,
payload.code,
)
except RuntimeError as exc:
if str(exc) == "rubika_protocol_not_connected":
raise HTTPException(
status_code=503,
detail="rubika_protocol_not_connected",
)
raise HTTPException(
status_code=400,
detail="invalid_code",
)
except Exception:
logger.exception(
"verify_code_failed"
)
raise HTTPException(
status_code=400,
detail="invalid_code",
)
# ----------------------------------------------------
# USER
# ----------------------------------------------------
phone = state.phone
user = (
await db.execute(
select(User).where(
User.phone == phone
)
)
).scalar_one_or_none()
if not user:
referral_id = payload.referral_id
user = User(
phone=phone,
referral_code=secrets.token_urlsafe(8),
is_banned=False,
)
db.add(user)
await db.flush()
if referral_id:
try:
referrer_id = int(
referral_id
)
if referrer_id != user.id:
referrer = await db.get(
User,
referrer_id,
)
if referrer:
existing_referral = (
await db.execute(
select(
ReferralUse.id
).where(
ReferralUse.user_id
== user.id
)
)
).scalar_one_or_none()
if not existing_referral:
db.add(
ReferralUse(
user_id=user.id,
referrer_id=referrer.id,
request_id=(
request.headers.get(
"X-Request-ID"
)
or str(
uuid.uuid4()
)
),
)
)
await add_coins(
db,
user.id,
100,
"referral_reward",
f"referral:{referrer.id}",
)
await add_coins(
db,
referrer.id,
100,
"referral_reward",
f"referral:{user.id}",
)
except (
ValueError,
TypeError,
):
pass
# ----------------------------------------------------
# ACCOUNT
# ----------------------------------------------------
account = (
await db.execute(
select(Account).where(
Account.user_id == user.id,
Account.phone == phone,
)
)
).scalar_one_or_none()
auth_value = result.get(
"auth",
"",
)
session_value = result.get(
"session",
"",
)
if account:
account.auth_data = encrypt_secret(
str(auth_value)
)
account.session_data = encrypt_secret(
str(session_value)
)
account.is_active = True
else:
account_count = (
await db.execute(
select(
func.count(Account.id)
).where(
Account.user_id
== user.id
)
)
).scalar_one()
account = Account(
user_id=user.id,
phone=phone,
auth_data=encrypt_secret(
str(auth_value)
),
session_data=encrypt_secret(
str(session_value)
),
is_active=True,
is_primary=(
account_count == 0
),
)
db.add(account)
await db.flush()
# ----------------------------------------------------
# TOKEN
# ----------------------------------------------------
raw_token = secrets.token_urlsafe(
48
)
token = ApiToken(
user_id=user.id,
account_id=account.id,
token_hash=hash_token(
raw_token
),
device_hash=device_hash,
expires_at=(
now_utc()
+ timedelta(
seconds=settings.TOKEN_TTL
)
),
is_active=True,
last_activity=now_utc(),
)
db.add(token)
state.user_id = user.id
state.status = "completed"
await db.commit()
return {
"ok": True,
"token": raw_token,
"expires_in": settings.TOKEN_TTL,
"user": {
"id": user.id,
"phone": user.phone,
},
"account": {
"id": account.id,
"phone": account.phone,
"is_primary": account.is_primary,
},
}
# ============================================================
# LOGOUT
# ============================================================
@app.post("/api/v1/auth/logout")
async def logout(
request: Request,
user: User = Depends(current_user),
authorization: Optional[str] = Header(
default=None
),
):
if not authorization:
return {
"ok": True
}
raw_token = authorization[7:].strip()
token_hash = hash_token(
raw_token
)
async with SessionLocal() as db:
token = (
await db.execute(
select(ApiToken).where(
ApiToken.token_hash
== token_hash,
ApiToken.user_id
== user.id,
)
)
).scalar_one_or_none()
if token:
token.is_active = False
await db.commit()
try:
await redis.set(
f"revoked-token:{token_hash}",
"1",
ex=settings.TOKEN_TTL,
)
except Exception:
pass
return {
"ok": True
}
# ============================================================
# ME
# ============================================================
@app.get("/api/v1/me")
async def me(
request: Request,
user: User = Depends(current_user),
):
async with SessionLocal() as db:
accounts = (
await db.execute(
select(Account).where(
Account.user_id == user.id,
Account.is_active == True,
)
)
).scalars().all()
balance = (
await db.execute(
select(
func.coalesce(
func.sum(
CoinTransaction.amount
),
0,
)
).where(
CoinTransaction.user_id
== user.id
)
)
).scalar_one()
return {
"ok": True,
"user": {
"id": user.id,
"phone": user.phone,
"referral_code": user.referral_code,
},
"coins": int(
balance or 0
),
"accounts": [
{
"id": a.id,
"phone": a.phone,
"is_primary": a.is_primary,
"is_active": a.is_active,
}
for a in accounts
],
}
# ============================================================
# ACCOUNTS
# ============================================================
@app.get("/api/v1/accounts")
async def accounts(
request: Request,
user: User = Depends(current_user),
):
async with SessionLocal() as db:
rows = (
await db.execute(
select(Account)
.where(
Account.user_id == user.id
)
.order_by(
Account.id.asc()
)
)
).scalars().all()
return {
"ok": True,
"items": [
{
"id": a.id,
"phone": a.phone,
"is_primary": a.is_primary,
"is_active": a.is_active,
}
for a in rows
],
}
@app.post(
"/api/v1/accounts/{account_id}/deactivate"
)
async def deactivate_account(
account_id: int,
request: Request,
user: User = Depends(current_user),
):
async with SessionLocal() as db:
account = await get_owned_account(
db,
user.id,
account_id,
)
account.is_active = False
await db.commit()
return {
"ok": True
}
# ============================================================
# COINS
# ============================================================
@app.get("/api/v1/coins")
async def coins(
request: Request,
user: User = Depends(current_user),
limit: int = 50,
offset: int = 0,
):
limit = max(
1,
min(limit, 100),
)
offset = max(
0,
offset,
)
async with SessionLocal() as db:
balance = (
await db.execute(
select(
func.coalesce(
func.sum(
CoinTransaction.amount
),
0,
)
).where(
CoinTransaction.user_id
== user.id
)
)
).scalar_one()
rows = (
await db.execute(
select(
CoinTransaction
)
.where(
CoinTransaction.user_id
== user.id
)
.order_by(
CoinTransaction.id.desc()
)
.limit(limit)
.offset(offset)
)
).scalars().all()
return {
"ok": True,
"balance": int(
balance or 0
),
"items": [
{
"id": x.id,
"amount": x.amount,
"type": x.type,
"reason": x.reason,
"created_at": (
x.created_at.isoformat()
if x.created_at
else None
),
}
for x in rows
],
}
# ============================================================
# COIN TRANSFER
# ============================================================
@app.post("/api/v1/coins/transfer")
async def transfer_coins(
payload: TransferRequest,
request: Request,
user: User = Depends(current_user),
):
request_id = request.headers.get(
"X-Request-ID"
)
if not request_id:
raise HTTPException(
status_code=400,
detail="request_id_required",
)
if payload.amount < settings.TRANSFER_MIN:
raise HTTPException(
status_code=400,
detail="transfer_amount_too_low",
)
if payload.target_user_id == user.id:
raise HTTPException(
status_code=400,
detail="cannot_transfer_to_self",
)
await rate_limit(
f"user:{user.id}:transfer",
20,
60,
)
async with SessionLocal() as db:
# Replay protection داخل transaction منطقی
duplicate = (
await db.execute(
select(
CoinTransaction.id
).where(
CoinTransaction.request_id
== request_id
)
)
).scalar_one_or_none()
if duplicate:
raise HTTPException(
status_code=409,
detail="duplicate_request",
)
sender = (
await db.execute(
select(User)
.where(
User.id == user.id
)
.with_for_update()
)
).scalar_one()
receiver = await db.get(
User,
payload.target_user_id,
)
if not receiver:
raise HTTPException(
status_code=404,
detail="target_user_not_found",
)
balance = (
await db.execute(
select(
func.coalesce(
func.sum(
CoinTransaction.amount
),
0,
)
).where(
CoinTransaction.user_id
== sender.id
)
)
).scalar_one()
balance = int(
balance or 0
)
tax = (
payload.amount
* settings.TRANSFER_TAX_PERCENT
+ 99
) // 100
total = (
payload.amount
+ tax
)
if balance < total:
raise HTTPException(
status_code=400,
detail="insufficient_coins",
)
db.add(
CoinTransaction(
user_id=sender.id,
amount=-total,
type="transfer_sent",
reason=(
f"to:{receiver.id}"
),
request_id=request_id,
)
)
db.add(
CoinTransaction(
user_id=receiver.id,
amount=payload.amount,
type="transfer_received",
reason=(
f"from:{sender.id}"
),
request_id=(
f"{request_id}:receiver"
),
)
)
await create_notification(
db,
receiver.id,
"دریافت سکه",
f"{payload.amount} سکه دریافت کردی.",
"coin_transfer",
)
await db.commit()
return {
"ok": True,
"amount": payload.amount,
"tax": tax,
"total": total,
}
# ============================================================
# ACCOUNT HELPER
# ============================================================
async def get_owned_account(
db: AsyncSession,
user_id: int,
account_id: int,
) -> Account:
account = (
await db.execute(
select(Account).where(
Account.id == account_id,
Account.user_id == user_id,
Account.is_active == True,
)
)
).scalar_one_or_none()
if not account:
raise HTTPException(
status_code=404,
detail="account_not_found",
)
return account
# ============================================================
# NOTIFICATION HELPER
# ============================================================
async def create_notification(
db: AsyncSession,
user_id: int,
title: str,
message: str,
kind: str = "system",
):
db.add(
Notification(
user_id=user_id,
title=title,
message=message,
kind=kind,
is_read=False,
)
)
# ============================================================
# RUBINO PROFILES
# ============================================================
@app.get(
"/api/v1/accounts/{account_id}/profiles"
)
async def account_profiles(
account_id: int,
request: Request,
user: User = Depends(current_user),
):
async with SessionLocal() as db:
account = await get_owned_account(
db,
user.id,
account_id,
)
try:
profiles = await rubika.profiles(
account=account
)
except RuntimeError as exc:
if str(exc) == "rubika_protocol_not_connected":
raise HTTPException(
status_code=503,
detail="rubika_protocol_not_connected",
)
raise HTTPException(
status_code=502,
detail="rubika_unavailable",
)
except Exception:
logger.exception(
"profiles_failed "
"user_id=%s account_id=%s",
user.id,
account.id,
)
raise HTTPException(
status_code=502,
detail="rubika_unavailable",
)
return {
"ok": True,
"profiles": profiles,
}
# ============================================================
# CREATE ORDER
# ============================================================
@app.post("/api/v1/orders")
async def create_order(
payload: CreateOrder,
request: Request,
user: User = Depends(current_user),
):
request_id = request.headers.get(
"X-Request-ID"
)
if not request_id:
raise HTTPException(
status_code=400,
detail="request_id_required",
)
if payload.followers < settings.MIN_ORDER:
raise HTTPException(
status_code=400,
detail=(
f"minimum_order_{settings.MIN_ORDER}"
),
)
await rate_limit(
f"user:{user.id}:create-order",
10,
60,
)
async with SessionLocal() as db:
duplicate = (
await db.execute(
select(
CoinTransaction.id
).where(
CoinTransaction.request_id
== request_id
)
)
).scalar_one_or_none()
if duplicate:
raise HTTPException(
status_code=409,
detail="duplicate_request",
)
account = await get_owned_account(
db,
user.id,
payload.account_id,
)
active_count = (
await db.execute(
select(
func.count(Order.id)
).where(
Order.account_id
== account.id,
Order.status.in_(
[
"pending",
"processing",
]
),
)
)
).scalar_one()
if active_count >= 10:
raise HTTPException(
status_code=429,
detail="too_many_active_orders",
)
total_cost = payload.followers
balance = (
await db.execute(
select(
func.coalesce(
func.sum(
CoinTransaction.amount
),
0,
)
).where(
CoinTransaction.user_id
== user.id
)
)
).scalar_one()
balance = int(
balance or 0
)
if balance < total_cost:
raise HTTPException(
status_code=400,
detail="insufficient_coins",
)
order = Order(
user_id=user.id,
account_id=account.id,
target_username=payload.username,
target_profile_id=payload.profile_id,
followers_required=payload.followers,
followers_done=0,
cost=total_cost,
status="pending",
)
db.add(order)
await db.flush()
db.add(
CoinTransaction(
user_id=user.id,
account_id=account.id,
amount=-total_cost,
type="order_create",
reason=f"order:{order.id}",
request_id=request_id,
)
)
await db.commit()
await db.refresh(order)
return {
"ok": True,
"order": {
"id": order.id,
"status": order.status,
"target_username": (
order.target_username
),
"target_profile_id": (
order.target_profile_id
),
"followers_required": (
order.followers_required
),
"followers_done": (
order.followers_done
),
"cost": order.cost,
},
}
# ============================================================
# MY ORDERS
# ============================================================
@app.get("/api/v1/orders/my")
async def my_orders(
request: Request,
user: User = Depends(current_user),
limit: int = 50,
offset: int = 0,
):
limit = max(
1,
min(limit, 100),
)
offset = max(
0,
offset,
)
async with SessionLocal() as db:
result = await db.execute(
select(Order)
.where(
Order.user_id == user.id
)
.order_by(
Order.id.desc()
)
.limit(limit)
.offset(offset)
)
orders = result.scalars().all()
return {
"ok": True,
"items": [
{
"id": o.id,
"target_username": (
o.target_username
),
"target_profile_id": (
o.target_profile_id
),
"followers_required": (
o.followers_required
),
"followers_done": (
o.followers_done
),
"cost": o.cost,
"status": o.status,
"created_at": (
o.created_at.isoformat()
if o.created_at
else None
),
}
for o in orders
],
}
# ============================================================
# AVAILABLE ORDERS
# ============================================================
@app.get("/api/v1/orders")
async def available_orders(
request: Request,
user: User = Depends(current_user),
limit: int = 50,
offset: int = 0,
):
limit = max(
1,
min(limit, 100),
)
offset = max(
0,
offset,
)
async with SessionLocal() as db:
result = await db.execute(
select(Order)
.where(
Order.status.in_(
[
"pending",
"processing",
]
),
Order.followers_done
< Order.followers_required,
Order.user_id != user.id,
)
.order_by(
Order.id.asc()
)
.limit(limit)
.offset(offset)
)
orders = result.scalars().all()
items = []
for order in orders:
already = (
await db.execute(
select(
OrderAction.id
).where(
OrderAction.order_id
== order.id,
OrderAction.user_id
== user.id,
).limit(1)
)
).scalar_one_or_none()
if already:
continue
items.append(
{
"id": order.id,
"target_username": (
order.target_username
),
"target_profile_id": (
order.target_profile_id
),
"followers_required": (
order.followers_required
),
"followers_done": (
order.followers_done
),
"remaining": (
order.followers_required
- order.followers_done
),
"reward": (
settings.FOLLOW_REWARD
),
}
)
return {
"ok": True,
"items": items,
}
# ============================================================
# FOLLOW ORDER
# ============================================================
@app.post(
"/api/v1/orders/{order_id}/follow"
)
async def follow_order(
order_id: int,
payload: FollowRequest,
request: Request,
user: User = Depends(current_user),
):
request_id = request.headers.get(
"X-Request-ID"
)
if not request_id:
raise HTTPException(
status_code=400,
detail="request_id_required",
)
await rate_limit(
f"user:{user.id}:follow",
60,
60,
)
async with SessionLocal() as db:
duplicate = (
await db.execute(
select(
CoinTransaction.id
).where(
CoinTransaction.request_id
== request_id
)
)
).scalar_one_or_none()
if duplicate:
raise HTTPException(
status_code=409,
detail="duplicate_request",
)
order = (
await db.execute(
select(Order)
.where(
Order.id == order_id
)
.with_for_update()
)
).scalar_one_or_none()
if not order:
raise HTTPException(
status_code=404,
detail="order_not_found",
)
if order.user_id == user.id:
raise HTTPException(
status_code=400,
detail="cannot_follow_own_order",
)
if order.status not in [
"pending",
"processing",
]:
raise HTTPException(
status_code=400,
detail="order_not_active",
)
if (
order.followers_done
>= order.followers_required
):
raise HTTPException(
status_code=400,
detail="order_completed",
)
existing = (
await db.execute(
select(OrderAction)
.where(
OrderAction.order_id
== order.id,
OrderAction.user_id
== user.id,
)
.with_for_update()
)
).scalar_one_or_none()
if existing:
raise HTTPException(
status_code=409,
detail="already_followed",
)
account = await get_owned_account(
db,
user.id,
payload.account_id,
)
lock_key = (
f"account-lock:{account.id}"
)
lock_value = secrets.token_hex(
16
)
acquired = await redis.set(
lock_key,
lock_value,
nx=True,
ex=60,
)
if not acquired:
raise HTTPException(
status_code=409,
detail="account_busy",
)
try:
try:
following = (
await rubika.is_following(
account=account,
profile_id=(
order.target_profile_id
),
)
)
except RuntimeError as exc:
if str(exc) == "rubika_protocol_not_connected":
raise HTTPException(
status_code=503,
detail="rubika_protocol_not_connected",
)
raise HTTPException(
status_code=502,
detail="rubika_unavailable",
)
except Exception:
logger.exception(
"rubika_follow_check_failed"
)
raise HTTPException(
status_code=502,
detail="rubika_unavailable",
)
if following:
raise HTTPException(
status_code=409,
detail="already_following",
)
try:
result = await rubika.follow(
account=account,
profile_id=(
order.target_profile_id
),
)
except RuntimeError as exc:
if str(exc) == "rubika_protocol_not_connected":
raise HTTPException(
status_code=503,
detail="rubika_protocol_not_connected",
)
raise HTTPException(
status_code=502,
detail="follow_failed",
)
except Exception:
logger.exception(
"rubika_follow_failed"
)
raise HTTPException(
status_code=502,
detail="follow_failed",
)
if result is False:
raise HTTPException(
status_code=400,
detail="follow_failed",
)
reward = settings.FOLLOW_REWARD
if account.is_primary:
reward += (
settings.PRIMARY_FOLLOW_BONUS
)
action = OrderAction(
order_id=order.id,
user_id=user.id,
account_id=account.id,
profile_id=(
order.target_profile_id
),
request_id=request_id,
reward=reward,
status="completed",
)
db.add(action)
order.followers_done += 1
if (
order.followers_done
>= order.followers_required
):
order.status = "completed"
else:
order.status = "processing"
db.add(
CoinTransaction(
user_id=user.id,
account_id=account.id,
amount=reward,
type="follow_reward",
reason=(
f"order:{order.id}"
),
request_id=request_id,
)
)
await create_notification(
db,
user.id,
"پاداش فالو",
(
f"{reward} سکه بابت "
"انجام فالو دریافت کردی."
),
"reward",
)
# Lucky Box
if (
settings.LUCKY_EVERY > 0
and order.followers_done
% settings.LUCKY_EVERY == 0
):
lucky_reward = (
secrets.randbelow(101)
)
if lucky_reward > 0:
lucky_request_id = (
f"{request_id}:lucky"
)
db.add(
CoinTransaction(
user_id=user.id,
account_id=account.id,
amount=lucky_reward,
type="lucky_box",
reason=(
f"order:{order.id}"
),
request_id=(
lucky_request_id
),
)
)
await create_notification(
db,
user.id,
"Lucky Box 🎁",
(
f"{lucky_reward} "
"سکه جایزه گرفتی."
),
"lucky_box",
)
if order.status == "completed":
await create_notification(
db,
order.user_id,
"سفارش کامل شد",
(
f"سفارش #{order.id} "
"کامل شد."
),
"order_completed",
)
await db.commit()
return {
"ok": True,
"message": "follow_completed",
"reward": reward,
"followers_done": (
order.followers_done
),
"followers_required": (
order.followers_required
),
"status": order.status,
}
finally:
try:
current_lock = await redis.get(
lock_key
)
if current_lock == lock_value:
await redis.delete(
lock_key
)
except Exception:
logger.warning(
"account_lock_release_failed"
)
# ============================================================
# DAILY REWARD
# ============================================================
@app.post("/api/v1/rewards/daily")
async def claim_daily_reward(
payload: ClaimRequest,
request: Request,
user: User = Depends(current_user),
):
request_id = request.headers.get(
"X-Request-ID"
)
if not request_id:
raise HTTPException(
status_code=400,
detail="request_id_required",
)
async with SessionLocal() as db:
today = (
datetime.now(
timezone.utc
)
.date()
.isoformat()
)
reason = (
f"daily:{today}"
)
existing = (
await db.execute(
select(
CoinTransaction.id
).where(
CoinTransaction.user_id
== user.id,
CoinTransaction.type
== "daily_reward",
CoinTransaction.reason
== reason,
).limit(1)
)
).scalar_one_or_none()
if existing:
raise HTTPException(
status_code=409,
detail=(
"daily_reward_already_claimed"
),
)
db.add(
CoinTransaction(
user_id=user.id,
account_id=payload.account_id,
amount=settings.DAILY_REWARD,
type="daily_reward",
reason=reason,
request_id=request_id,
)
)
await create_notification(
db,
user.id,
"پاداش روزانه 🎁",
(
f"{settings.DAILY_REWARD} "
"سکه دریافت کردی."
),
"daily_reward",
)
await db.commit()
return {
"ok": True,
"reward": settings.DAILY_REWARD,
"date": today,
}
# ============================================================
# NOTIFICATIONS
# ============================================================
@app.get("/api/v1/notifications")
async def notifications(
request: Request,
user: User = Depends(current_user),
limit: int = 50,
offset: int = 0,
):
limit = max(
1,
min(limit, 100),
)
offset = max(
0,
offset,
)
async with SessionLocal() as db:
result = await db.execute(
select(Notification)
.where(
Notification.user_id
== user.id
)
.order_by(
Notification.id.desc()
)
.limit(limit)
.offset(offset)
)
rows = result.scalars().all()
return {
"ok": True,
"items": [
{
"id": n.id,
"title": n.title,
"message": n.message,
"kind": n.kind,
"is_read": n.is_read,
"created_at": (
n.created_at.isoformat()
if n.created_at
else None
),
}
for n in rows
],
}
@app.post(
"/api/v1/notifications/{notification_id}/read"
)
async def read_notification(
notification_id: int,
request: Request,
user: User = Depends(current_user),
):
async with SessionLocal() as db:
notification = (
await db.execute(
select(Notification).where(
Notification.id
== notification_id,
Notification.user_id
== user.id,
)
)
).scalar_one_or_none()
if not notification:
raise HTTPException(
status_code=404,
detail="notification_not_found",
)
notification.is_read = True
await db.commit()
return {
"ok": True
}
# ============================================================
# REFERRAL
# ============================================================
@app.post("/api/v1/referral/use")
async def use_referral(
payload: ReferralRequest,
request: Request,
user: User = Depends(current_user),
):
request_id = request.headers.get(
"X-Request-ID"
)
if not request_id:
raise HTTPException(
status_code=400,
detail="request_id_required",
)
if payload.referrer_id == user.id:
raise HTTPException(
status_code=400,
detail="cannot_refer_self",
)
async with SessionLocal() as db:
existing = (
await db.execute(
select(ReferralUse).where(
ReferralUse.user_id
== user.id
)
)
).scalar_one_or_none()
if existing:
raise HTTPException(
status_code=409,
detail="referral_already_used",
)
referrer = await db.get(
User,
payload.referrer_id,
)
if not referrer:
raise HTTPException(
status_code=404,
detail="referrer_not_found",
)
db.add(
ReferralUse(
user_id=user.id,
referrer_id=referrer.id,
request_id=request_id,
)
)
await add_coins(
db,
user.id,
100,
"referral_reward",
f"referral:{referrer.id}",
request_id=f"{request_id}:user",
)
await add_coins(
db,
referrer.id,
100,
"referral_reward",
f"referral:{user.id}",
request_id=(
f"{request_id}:referrer"
),
)
await create_notification(
db,
user.id,
"پاداش دعوت 🎉",
(
"100 سکه بابت استفاده "
"از کد دعوت دریافت کردی."
),
"referral",
)
await create_notification(
db,
referrer.id,
"دعوت موفق 🎉",
(
"100 سکه بابت دعوت "
"یک کاربر جدید دریافت کردی."
),
"referral",
)
await db.commit()
return {
"ok": True,
"reward": 100,
}
# ============================================================
# SUPPORT
# ============================================================
@app.post("/api/v1/support/tickets")
async def create_support_ticket(
payload: TicketRequest,
request: Request,
user: User = Depends(current_user),
):
await rate_limit(
f"support:{user.id}",
settings.SUPPORT_DAILY_LIMIT,
86400,
)
async with SessionLocal() as db:
ticket = SupportTicket(
user_id=user.id,
subject=payload.subject,
status="open",
)
db.add(ticket)
await db.flush()
db.add(
SupportMessage(
ticket_id=ticket.id,
user_id=user.id,
message=payload.message,
)
)
await db.commit()
await db.refresh(ticket)
return {
"ok": True,
"ticket_id": ticket.id,
"status": ticket.status,
}
@app.post(
"/api/v1/support/tickets/{ticket_id}/messages"
)
async def add_support_message(
ticket_id: int,
payload: MessageRequest,
request: Request,
user: User = Depends(current_user),
):
async with SessionLocal() as db:
ticket = (
await db.execute(
select(SupportTicket).where(
SupportTicket.id
== ticket_id,
SupportTicket.user_id
== user.id,
)
)
).scalar_one_or_none()
if not ticket:
raise HTTPException(
status_code=404,
detail="ticket_not_found",
)
if ticket.status == "closed":
raise HTTPException(
status_code=400,
detail="ticket_closed",
)
db.add(
SupportMessage(
ticket_id=ticket.id,
user_id=user.id,
message=payload.message,
)
)
await db.commit()
return {
"ok": True,
"message": "message_sent",
}
@app.get("/api/v1/support/tickets")
async def support_tickets(
request: Request,
user: User = Depends(current_user),
):
async with SessionLocal() as db:
result = await db.execute(
select(SupportTicket)
.where(
SupportTicket.user_id
== user.id
)
.order_by(
SupportTicket.id.desc()
)
)
tickets = result.scalars().all()
return {
"ok": True,
"items": [
{
"id": t.id,
"subject": t.subject,
"status": t.status,
"created_at": (
t.created_at.isoformat()
if t.created_at
else None
),
}
for t in tickets
],
}
# ============================================================
# SECTION 3
# UNFOLLOW + DEBT + LEADERBOARD + LUCKY BOX
# REPORTS + FRAUD/RISK + ADMIN
# ============================================================
import random
# ============================================================
# EXTRA MODELS
# ============================================================
class Report(Base):
__tablename__ = "reports"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
reporter_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
reported_user_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("users.id"),
nullable=True,
index=True,
)
order_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("orders.id"),
nullable=True,
index=True,
)
reason: Mapped[str] = mapped_column(String(100))
description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
status: Mapped[str] = mapped_column(
String(30),
default="open",
index=True,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
index=True,
)
resolved_at: Mapped[Optional[datetime]] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
class LuckyBox(Base):
__tablename__ = "lucky_boxes"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id"),
index=True,
)
order_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("orders.id"),
nullable=True,
index=True,
)
reward: Mapped[int] = mapped_column(Integer, default=0)
claimed: Mapped[bool] = mapped_column(Boolean, default=False, index=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
index=True,
)
claimed_at: Mapped[Optional[datetime]] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
class RiskEvent(Base):
__tablename__ = "risk_events"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("users.id"),
nullable=True,
index=True,
)
account_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("accounts.id"),
nullable=True,
index=True,
)
event_type: Mapped[str] = mapped_column(String(80), index=True)
severity: Mapped[str] = mapped_column(String(20), default="low")
score: Mapped[int] = mapped_column(Integer, default=0)
ip_hash: Mapped[Optional[str]] = mapped_column(
String(128),
nullable=True,
index=True,
)
device_hash: Mapped[Optional[str]] = mapped_column(
String(128),
nullable=True,
index=True,
)
details: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
index=True,
)
class LeaderboardSnapshot(Base):
__tablename__ = "leaderboard_snapshots"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
period: Mapped[str] = mapped_column(String(30), index=True)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id"),
index=True,
)
rank: Mapped[int] = mapped_column(Integer)
score: Mapped[int] = mapped_column(BigInteger, default=0)
reward: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=now_utc,
)
# ============================================================
# SCHEMAS
# ============================================================
class ReportRequest(BaseModel):
request_id: str = Field(min_length=8, max_length=128)
reported_user_id: Optional[int] = None
order_id: Optional[int] = None
reason: str = Field(min_length=2, max_length=100)
description: Optional[str] = Field(default=None, max_length=2000)
class LuckyBoxClaimRequest(BaseModel):
request_id: str = Field(min_length=8, max_length=128)
class AdminReportAction(BaseModel):
action: str = Field(pattern="^(resolve|reject|warn|ban)$")
note: Optional[str] = Field(default=None, max_length=2000)
class AdminUserAction(BaseModel):
action: str = Field(
pattern="^(ban|unban|suspend|unsuspend|add_coins|remove_coins)$"
)
amount: int = Field(default=0, ge=0, le=10_000_000)
reason: Optional[str] = Field(default=None, max_length=1000)
# ============================================================
# HELPERS
# ============================================================
async def create_notification(
db: AsyncSession,
user_id: int,
title: str,
body: str,
kind: str = "system",
):
notification = Notification(
user_id=user_id,
title=title,
body=body,
kind=kind,
is_read=False,
created_at=now_utc(),
)
db.add(notification)
return notification
async def add_risk_event(
db: AsyncSession,
*,
user_id: Optional[int],
account_id: Optional[int],
event_type: str,
severity: str,
score: int,
ip_hash: Optional[str] = None,
device_hash: Optional[str] = None,
details: Optional[str] = None,
):
event = RiskEvent(
user_id=user_id,
account_id=account_id,
event_type=event_type,
severity=severity,
score=score,
ip_hash=ip_hash,
device_hash=device_hash,
details=details,
created_at=now_utc(),
)
db.add(event)
return event
async def get_coin_balance(
db: AsyncSession,
user_id: int,
) -> int:
result = await db.execute(
select(
func.coalesce(
func.sum(CoinTransaction.amount),
0,
)
).where(
CoinTransaction.user_id == user_id
)
)
return int(result.scalar_one() or 0)
async def add_debt(
db: AsyncSession,
user_id: int,
amount: int,
reason: str,
request_id: str,
):
if amount <= 0:
return
existing = await db.execute(
select(CoinTransaction.id).where(
CoinTransaction.request_id == request_id
)
)
if existing.scalar_one_or_none() is not None:
return
tx = CoinTransaction(
user_id=user_id,
amount=-abs(amount),
reason=reason,
request_id=request_id,
created_at=now_utc(),
)
db.add(tx)
async def calculate_leaderboard(
db: AsyncSession,
limit: int = 10,
):
result = await db.execute(
select(
CoinTransaction.user_id,
func.coalesce(
func.sum(CoinTransaction.amount),
0,
).label("score"),
)
.group_by(CoinTransaction.user_id)
.order_by(
func.sum(CoinTransaction.amount).desc()
)
.limit(limit)
)
rows = result.all()
output = []
for rank, row in enumerate(rows, start=1):
output.append(
{
"rank": rank,
"user_id": row.user_id,
"score": int(row.score or 0),
}
)
return output
# ============================================================
# UNFOLLOW DETECTION
# ============================================================
@app.post("/api/v1/accounts/{account_id}/check-follow/{order_id}")
async def check_follow_status(
account_id: int,
order_id: int,
request: Request,
db: AsyncSession = Depends(SessionLocal),
user: User = Depends(current_user),
):
account = await get_owned_account(
db,
user.id,
account_id,
)
if account is None:
raise HTTPException(
status_code=404,
detail="account_not_found",
)
order_result = await db.execute(
select(Order).where(Order.id == order_id)
)
order = order_result.scalar_one_or_none()
if order is None:
raise HTTPException(
status_code=404,
detail="order_not_found",
)
action_result = await db.execute(
select(OrderAction).where(
OrderAction.order_id == order_id,
OrderAction.user_id == user.id,
)
)
action = action_result.scalar_one_or_none()
if action is None:
raise HTTPException(
status_code=404,
detail="follow_action_not_found",
)
try:
following = await rubika.is_following(
account,
order.username,
)
except Exception:
raise HTTPException(
status_code=503,
detail="rubika_service_unavailable",
)
if following:
return {
"ok": True,
"following": True,
"penalty": 0,
}
# --------------------------------------------------------
# UNFOLLOW DETECTED
# --------------------------------------------------------
existing = await db.execute(
select(Unfollow).where(
Unfollow.order_action_id == action.id
)
)
unfollow = existing.scalar_one_or_none()
if unfollow is not None:
return {
"ok": True,
"following": False,
"already_recorded": True,
"penalty": settings.UNFOLLOW_PENALTY,
}
unfollow = Unfollow(
user_id=user.id,
order_id=order.id,
order_action_id=action.id,
penalty=settings.UNFOLLOW_PENALTY,
created_at=now_utc(),
)
db.add(unfollow)
await add_debt(
db,
user.id,
settings.UNFOLLOW_PENALTY,
"unfollow_penalty",
f"unfollow:{action.id}",
)
await create_notification(
db,
user.id,
"Unfollow detected",
(
f"You unfollowed an order target. "
f"{settings.UNFOLLOW_PENALTY} coins were added as debt."
),
"unfollow",
)
await add_risk_event(
db,
user_id=user.id,
account_id=account.id,
event_type="unfollow",
severity="medium",
score=10,
ip_hash=hash_token(get_client_ip(request)),
device_hash=get_device_hash(request),
details=json.dumps(
{
"order_id": order.id,
"action_id": action.id,
}
),
)
await db.commit()
return {
"ok": True,
"following": False,
"penalty": settings.UNFOLLOW_PENALTY,
"debt_added": True,
}
# ============================================================
# BULK UNFOLLOW CHECK
# ============================================================
@app.post("/api/v1/accounts/{account_id}/check-follows")
async def check_all_follows(
account_id: int,
request: Request,
db: AsyncSession = Depends(SessionLocal),
user: User = Depends(current_user),
):
account = await get_owned_account(
db,
user.id,
account_id,
)
if account is None:
raise HTTPException(
status_code=404,
detail="account_not_found",
)
result = await db.execute(
select(OrderAction, Order)
.join(
Order,
Order.id == OrderAction.order_id,
)
.where(
OrderAction.user_id == user.id,
OrderAction.account_id == account.id,
)
.order_by(
OrderAction.created_at.desc()
)
.limit(100)
)
rows = result.all()
checked = 0
unfollowed = 0
for action, order in rows:
checked += 1
try:
following = await rubika.is_following(
account,
order.username,
)
except Exception:
continue
if following:
continue
existing = await db.execute(
select(Unfollow).where(
Unfollow.order_action_id == action.id
)
)
if existing.scalar_one_or_none() is not None:
continue
unfollow = Unfollow(
user_id=user.id,
order_id=order.id,
order_action_id=action.id,
penalty=settings.UNFOLLOW_PENALTY,
created_at=now_utc(),
)
db.add(unfollow)
await add_debt(
db,
user.id,
settings.UNFOLLOW_PENALTY,
"unfollow_penalty",
f"unfollow:{action.id}",
)
unfollowed += 1
if unfollowed:
await create_notification(
db,
user.id,
"Unfollow check completed",
f"{unfollowed} unfollow(s) detected.",
"unfollow",
)
await db.commit()
return {
"ok": True,
"checked": checked,
"unfollowed": unfollowed,
}
# ============================================================
# DEBT STATUS
# ============================================================
@app.get("/api/v1/coins/debt")
async def get_debt(
db: AsyncSession = Depends(SessionLocal),
user: User = Depends(current_user),
):
result = await db.execute(
select(
func.coalesce(
func.sum(CoinTransaction.amount),
0,
)
).where(
CoinTransaction.user_id == user.id,
CoinTransaction.amount < 0,
CoinTransaction.reason.like("%penalty%"),
)
)
total_penalties = abs(
int(result.scalar_one() or 0)
)
balance = await get_coin_balance(
db,
user.id,
)
return {
"ok": True,
"balance": balance,
"penalty_debt": total_penalties,
}
# ============================================================
# LEADERBOARD
# ============================================================
@app.get("/api/v1/leaderboard")
async def leaderboard(
limit: int = 10,
db: AsyncSession = Depends(SessionLocal),
user: User = Depends(current_user),
):
limit = max(1, min(limit, 10))
rows = await calculate_leaderboard(
db,
limit,
)
return {
"ok": True,
"period": "all_time",
"items": rows,
}
@app.get("/api/v1/leaderboard/me")
async def leaderboard_me(
db: AsyncSession = Depends(SessionLocal),
user: User = Depends(current_user),
):
rows = await calculate_leaderboard(
db,
1000,
)
rank = None
score = 0
for item in rows:
if item["user_id"] == user.id:
rank = item["rank"]
score = item["score"]
break
return {
"ok": True,
"rank": rank,
"score": score,
}
# ============================================================
# LUCKY BOX
# ============================================================
@app.get("/api/v1/rewards/lucky-boxes")
async def lucky_boxes(
db: AsyncSession = Depends(SessionLocal),
user: User = Depends(current_user),
):
result = await db.execute(
select(LuckyBox)
.where(
LuckyBox.user_id == user.id
)
.order_by(
LuckyBox.created_at.desc()
)
.limit(50)
)
boxes = result.scalars().all()
return {
"ok": True,
"items": [
{
"id": box.id,
"reward": box.reward if box.claimed else None,
"claimed": box.claimed,
"created_at": box.created_at.isoformat(),
}
for box in boxes
],
}
@app.post("/api/v1/rewards/lucky-box/{box_id}/claim")
async def claim_lucky_box(
box_id: int,
payload: LuckyBoxClaimRequest,
db: AsyncSession = Depends(SessionLocal),
user: User = Depends(current_user),
):
if not await consume_request_id(
payload.request_id,
ttl=settings.REQUEST_WINDOW,
):
raise HTTPException(
status_code=409,
detail="duplicate_request",
)
result = await db.execute(
select(LuckyBox)
.where(
LuckyBox.id == box_id,
LuckyBox.user_id == user.id,
)
.with_for_update()
)
box = result.scalar_one_or_none()
if box is None:
raise HTTPException(
status_code=404,
detail="lucky_box_not_found",
)
if box.claimed:
raise HTTPException(
status_code=409,
detail="lucky_box_already_claimed",
)
reward = max(
0,
min(
int(box.reward),
100,
),
)
box.claimed = True
box.claimed_at = now_utc()
if reward > 0:
tx = CoinTransaction(
user_id=user.id,
amount=reward,
reason="lucky_box",
request_id=payload.request_id,
created_at=now_utc(),
)
db.add(tx)
await create_notification(
db,
user.id,
"Lucky Box",
f"You received {reward} coins.",
"lucky_box",
)
await db.commit()
return {
"ok": True,
"reward": reward,
"claimed": True,
}
# ============================================================
# REPORTS
# ============================================================
@app.post("/api/v1/reports")
async def create_report(
payload: ReportRequest,
request: Request,
db: AsyncSession = Depends(SessionLocal),
user: User = Depends(current_user),
):
if not await consume_request_id(
payload.request_id,
ttl=settings.REQUEST_WINDOW,
):
raise HTTPException(
status_code=409,
detail="duplicate_request",
)
await rate_limit(
f"report:{user.id}",
10,
86400,
)
if payload.reported_user_id == user.id:
raise HTTPException(
status_code=400,
detail="cannot_report_self",
)
if payload.order_id is not None:
order_result = await db.execute(
select(Order).where(
Order.id == payload.order_id
)
)
if order_result.scalar_one_or_none() is None:
raise HTTPException(
status_code=404,
detail="order_not_found",
)
report = Report(
reporter_id=user.id,
reported_user_id=payload.reported_user_id,
order_id=payload.order_id,
reason=payload.reason,
description=payload.description,
status="open",
created_at=now_utc(),
)
db.add(report)
await add_risk_event(
db,
user_id=payload.reported_user_id,
account_id=None,
event_type="report_received",
severity="medium",
score=5,
ip_hash=hash_token(get_client_ip(request)),
device_hash=get_device_hash(request),
details=json.dumps(
{
"reporter_id": user.id,
"reason": payload.reason,
"order_id": payload.order_id,
}
),
)
await db.commit()
return {
"ok": True,
"report_id": report.id,
"status": report.status,
}
@app.get("/api/v1/reports/my")
async def my_reports(
db: AsyncSession = Depends(SessionLocal),
user: User = Depends(current_user),
):
result = await db.execute(
select(Report)
.where(
Report.reporter_id == user.id
)
.order_by(
Report.created_at.desc()
)
.limit(100)
)
reports = result.scalars().all
# ============================================================
# SECTION 4
# WORKER + QUEUE + RETRY + BACKOFF + DLQ
# GRACEFUL SHUTDOWN + API/WORKER ENTRYPOINT
# ============================================================
import sys
import socket
import traceback
# ============================================================
# WORKER CONFIG
# ============================================================
WORKER_ID = (
f"{socket.gethostname()}:"
f"{os.getpid()}:"
f"{uuid.uuid4().hex[:8]}"
)
WORKER_POLL_INTERVAL = 1.0
WORKER_BATCH_SIZE = 10
WORKER_MAX_RETRIES = 5
WORKER_RETRY_BASE = 2
WORKER_RETRY_MAX = 300
_shutdown_event = asyncio.Event()
# ============================================================
# JOB STATUS CONSTANTS
# ============================================================
JOB_PENDING = "pending"
JOB_PROCESSING = "processing"
JOB_COMPLETED = "completed"
JOB_FAILED = "failed"
JOB_DLQ = "dlq"
# ============================================================
# SAFE JSON
# ============================================================
def safe_json(value):
try:
return json.dumps(
value,
ensure_ascii=False,
separators=(",", ":"),
)
except Exception:
return "{}"
# ============================================================
# JOB LOCK
# ============================================================
async def acquire_job_lock(job_id: int):
key = f"job-lock:{job_id}"
token = secrets.token_hex(16)
acquired = await redis.set(
key,
token,
nx=True,
ex=300,
)
if not acquired:
return None
return token
async def release_job_lock(
job_id: int,
token: Optional[str],
):
if not token:
return
key = f"job-lock:{job_id}"
current = await redis.get(key)
if current == token:
await redis.delete(key)
# ============================================================
# WORKER INSTANCE LOCK
# ============================================================
async def register_worker():
key = f"worker:{WORKER_ID}"
await redis.set(
key,
safe_json(
{
"worker_id": WORKER_ID,
"hostname": socket.gethostname(),
"pid": os.getpid(),
"started_at": now_utc().isoformat(),
}
),
ex=30,
)
async def worker_heartbeat():
key = f"worker:{WORKER_ID}"
while not _shutdown_event.is_set():
try:
await redis.expire(
key,
30,
)
except Exception:
pass
try:
await asyncio.wait_for(
_shutdown_event.wait(),
timeout=10,
)
except asyncio.TimeoutError:
pass
async def unregister_worker():
try:
await redis.delete(
f"worker:{WORKER_ID}"
)
except Exception:
pass
# ============================================================
# RETRY CALCULATION
# ============================================================
def calculate_retry_delay(
retry_count: int,
) -> int:
retry_count = max(
0,
min(
retry_count,
WORKER_MAX_RETRIES,
),
)
delay = (
WORKER_RETRY_BASE
** retry_count
)
delay = min(
delay,
WORKER_RETRY_MAX,
)
jitter = random.randint(
0,
max(1, delay // 4),
)
return min(
WORKER_RETRY_MAX,
delay + jitter,
)
# ============================================================
# CLAIM NEXT JOB
# ============================================================
async def claim_next_job(
db: AsyncSession,
):
result = await db.execute(
select(Job)
.where(
Job.status == JOB_PENDING
)
.order_by(
Job.priority.desc(),
Job.created_at.asc(),
)
.with_for_update(
skip_locked=True
)
.limit(1)
)
job = result.scalar_one_or_none()
if job is None:
return None
job.status = JOB_PROCESSING
# These fields may exist in older schemas only.
# We intentionally do not assume extra Job columns.
await db.commit()
return job
# ============================================================
# JOB FAILURE HANDLER
# ============================================================
async def mark_job_failure(
db: AsyncSession,
job: Job,
error: Exception,
):
error_text = str(error)
if len(error_text) > 2000:
error_text = error_text[:2000]
# Try to read retry_count from the model if present.
retry_count = getattr(
job,
"retry_count",
0,
)
try:
retry_count = int(
retry_count or 0
)
except Exception:
retry_count = 0
retry_count += 1
# --------------------------------------------------------
# Retry
# --------------------------------------------------------
if retry_count <= WORKER_MAX_RETRIES:
if hasattr(job, "retry_count"):
job.retry_count = retry_count
job.status = JOB_PENDING
# Store retry metadata inside payload.
try:
payload = json.loads(
job.payload or "{}"
)
except Exception:
payload = {}
payload["_worker_retry_count"] = retry_count
payload["_last_error"] = error_text
payload["_next_retry_at"] = (
datetime.now(
timezone.utc
)
+ timedelta(
seconds=calculate_retry_delay(
retry_count
)
)
).isoformat()
job.payload = safe_json(
payload
)
await db.commit()
return "retry"
# --------------------------------------------------------
# Dead Letter Queue
# --------------------------------------------------------
job.status = JOB_DLQ
try:
payload = json.loads(
job.payload or "{}"
)
except Exception:
payload = {}
payload["_worker_retry_count"] = retry_count
payload["_last_error"] = error_text
payload["_failed_at"] = (
now_utc().isoformat()
)
payload["_worker_id"] = WORKER_ID
job.payload = safe_json(
payload
)
await db.commit()
return "dlq"
# ============================================================
# JOB SUCCESS HANDLER
# ============================================================
async def mark_job_completed(
db: AsyncSession,
job: Job,
):
job.status = JOB_COMPLETED
await db.commit()
# ============================================================
# JOB DISPATCHER
# ============================================================
async def process_job(
job: Job,
):
"""
Dispatches a Job according to its type.
Supported job types:
follow
unfollow_check
notification
generic
"""
try:
payload = json.loads(
job.payload or "{}"
)
except Exception:
payload = {}
job_type = getattr(
job,
"job_type",
None,
)
# Some earlier schemas may call it type.
if not job_type:
job_type = getattr(
job,
"type",
None,
)
# --------------------------------------------------------
# FOLLOW JOB
# --------------------------------------------------------
if job_type == "follow":
account_id = payload.get(
"account_id"
)
order_id = payload.get(
"order_id"
)
user_id = payload.get(
"user_id"
)
if not account_id or not order_id:
raise RuntimeError(
"invalid_follow_job"
)
async with SessionLocal() as db:
account_result = await db.execute(
select(Account).where(
Account.id == account_id
)
)
account = (
account_result
.scalar_one_or_none()
)
if account is None:
raise RuntimeError(
"account_not_found"
)
order_result = await db.execute(
select(Order).where(
Order.id == order_id
)
)
order = (
order_result
.scalar_one_or_none()
)
if order is None:
raise RuntimeError(
"order_not_found"
)
if order.status == "completed":
return
try:
following = (
await rubika.is_following(
account,
order.username,
)
)
if not following:
await rubika.follow(
account,
order.username,
)
except Exception as exc:
raise RuntimeError(
f"rubika_follow_failed: "
f"{str(exc)[:500]}"
) from exc
# ------------------------------------------------
# Atomic-ish order update
# ------------------------------------------------
order.followers_done = (
int(order.followers_done or 0)
+ 1
)
if (
order.followers_done
>= order.followers
):
order.followers_done = (
order.followers
)
order.status = "completed"
else:
order.status = "processing"
# ------------------------------------------------
# Reward
# ------------------------------------------------
reward = settings.FOLLOW_REWARD
if getattr(
account,
"is_primary",
False,
):
reward += (
settings.PRIMARY_FOLLOW_BONUS
)
# ------------------------------------------------
# Prevent duplicate reward
# ------------------------------------------------
reward_request_id = (
f"worker-follow:"
f"{order.id}:"
f"{user_id}:"
f"{account.id}"
)
duplicate = await db.execute(
select(
CoinTransaction.id
).where(
CoinTransaction.request_id
== reward_request_id
)
)
if (
duplicate.scalar_one_or_none()
is None
):
db.add(
CoinTransaction(
user_id=user_id,
amount=reward,
reason="follow_reward",
request_id=(
reward_request_id
),
created_at=now_utc(),
)
)
await db.commit()
return
# --------------------------------------------------------
# UNFOLLOW CHECK
# --------------------------------------------------------
if job_type == "unfollow_check":
account_id = payload.get(
"account_id"
)
user_id = payload.get(
"user_id"
)
if not account_id or not user_id:
raise RuntimeError(
"invalid_unfollow_job"
)
async with SessionLocal() as db:
account_result = await db.execute(
select(Account).where(
Account.id == account_id,
Account.user_id == user_id,
)
)
account = (
account_result
.scalar_one_or_none()
)
if account is None:
raise RuntimeError(
"account_not_found"
)
result = await db.execute(
select(
OrderAction,
Order,
)
.join(
Order,
Order.id
== OrderAction.order_id,
)
.where(
OrderAction.user_id
== user_id,
OrderAction.account_id
== account_id,
)
.order_by(
OrderAction.created_at.desc()
)
.limit(100)
)
rows = result.all()
for action, order in rows:
try:
following = (
await rubika.is_following(
account,
order.username,
)
)
except Exception:
continue
if following:
continue
existing = await db.execute(
select(Unfollow).where(
Unfollow.order_action_id
== action.id
)
)
if (
existing.scalar_one_or_none()
is not None
):
continue
db.add(
Unfollow(
user_id=user_id,
order_id=order.id,
order_action_id=action.id,
penalty=(
settings.UNFOLLOW_PENALTY
),
created_at=now_utc(),
)
)
penalty_request = (
f"worker-unfollow:"
f"{action.id}"
)
duplicate = await db.execute(
select(
CoinTransaction.id
).where(
CoinTransaction.request_id
== penalty_request
)
)
if (
duplicate.scalar_one_or_none()
is None
):
db.add(
CoinTransaction(
user_id=user_id,
amount=-abs(
settings.UNFOLLOW_PENALTY
),
reason=(
"unfollow_penalty"
),
request_id=(
penalty_request
),
created_at=now_utc(),
)
)
await db.commit()
return
# --------------------------------------------------------
# NOTIFICATION JOB
# --------------------------------------------------------
if job_type == "notification":
user_id = payload.get(
"user_id"
)
title = payload.get(
"title",
"Notification",
)
body = payload.get(
"body",
"",
)
if not user_id:
raise RuntimeError(
"invalid_notification_job"
)
async with SessionLocal() as db:
await create_notification(
db,
user_id,
title,
body,
"worker",
)
await db.commit()
return
# --------------------------------------------------------
# GENERIC JOB
# --------------------------------------------------------
if job_type in (
None,
"",
"generic",
):
# Unknown generic jobs should not
# silently disappear.
raise RuntimeError(
"unsupported_job_type"
)
raise RuntimeError(
f"unsupported_job_type:{job_type}"
)
# ============================================================
# PROCESS ONE JOB
# ============================================================
async def process_one_job(
job_id: int,
):
lock_token = await acquire_job_lock(
job_id
)
if not lock_token:
return
try:
async with SessionLocal() as db:
result = await db.execute(
select(Job).where(
Job.id == job_id
)
)
job = (
result.scalar_one_or_none()
)
if job is None:
return
if job.status != JOB_PROCESSING:
return
try:
await process_job(
job
)
except Exception as exc:
await mark_job_failure(
db,
job,
exc,
)
logging.exception(
"Worker job failed: %s",
job_id,
)
return
# Reload because process_job
# uses its own DB session.
result = await db.execute(
select(Job).where(
Job.id == job_id
)
)
job = (
result.scalar_one_or_none()
)
if job is not None:
await mark_job_completed(
db,
job,
)
finally:
await release_job_lock(
job_id,
lock_token,
)
# ============================================================
# RECOVER STALE JOBS
# ============================================================
async def recover_stale_jobs():
"""
Recover jobs that were stuck in processing
because a worker crashed.
Since the original Job model may not contain
started_at/heartbeat columns, recovery is done
conservatively through Redis worker locks.
"""
try:
keys = []
async for key in redis.scan_iter(
match="job-lock:*",
count=100,
):
keys.append(key)
locked_ids = set()
for key in keys:
try:
job_id = int(
key.split(":")[-1]
)
locked_ids.add(job_id)
except Exception:
continue
async with SessionLocal() as db:
result = await db.execute(
select(Job).where(
Job.status
== JOB_PROCESSING
)
)
jobs = result.scalars().all()
changed = 0
for job in jobs:
if job.id in locked_ids:
continue
job.status = JOB_PENDING
changed += 1
if changed:
await db.commit()
except Exception:
logging.exception(
"Failed to recover stale jobs"
)
# ============================================================
# QUEUE STATS
# ============================================================
a$ tail -f comments.log
Comments [0]
// no comments yet