# 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:postgres@127.0.0.1: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