feat: Yandex Metrika offline conversions + S2S postbacks

- Add YandexClientIdMap model for user → yandex_cid mapping with
  upsert-safe CRUD (ON CONFLICT DO UPDATE)
- Add yandex_cid, subid, referrer columns to GuestPurchase
- Add yandex_offline_conv_service: Measurement Protocol integration
  with mc.yandex.ru/collect (registration, trial, purchase events),
  background task management, CID parsing from /start params
- Add s2s_postback_service: server-to-server affiliate postbacks
  with URL template placeholders and URL-safe encoding
- Add analytics offline conversion info to branding API (masked secret)
- Add POST /analytics/yandex-cid endpoint for cabinet CID capture
- Add 11 config settings (YANDEX_OFFLINE_CONV_*, S2S_POSTBACK_*)
- Add migration 0063 (yandex_client_id_map table + guest_purchases cols)
- Fix: mask measurement secret aggressively (show only last 4 chars)
- Fix: always replace {user_id} placeholder in S2S postback URLs
- Fix: use structlog kwargs instead of f-strings with LOG_PREFIX

Based on PR #2851 by @smediainfo — CI/CD workflow changes excluded

app/cabinet/routes/branding.py

@@ -17,7 +17,7 @@ from app.config import settings
 from app.database.crud.system_setting import get_setting_value
 from app.database.models import SystemSetting, User
 
-from ..dependencies import get_cabinet_db, require_permission
+from ..dependencies import get_cabinet_db, get_current_cabinet_user, require_permission
 
 
 logger = structlog.get_logger(__name__)
@@ -291,12 +291,24 @@ class GiftEnabledUpdate(BaseModel):
     enabled: bool
 
 
+class OfflineConvGoal(BaseModel):
+    """Yandex Metrika offline conversion goal descriptor."""
+
+    name: str
+    event_id: str
+    dedup: str
+
+
 class AnalyticsCountersResponse(BaseModel):
     """Analytics counter settings."""
 
     yandex_metrika_id: str = ''
     google_ads_id: str = ''
     google_ads_label: str = ''
+    offline_conv_enabled: bool = False
+    offline_conv_counter_id: str = ''
+    offline_conv_measurement_secret_masked: str = ''
+    offline_conv_goals: list[OfflineConvGoal] = []
 
 
 class AnalyticsCountersUpdate(BaseModel):
@@ -924,10 +936,27 @@ async def get_analytics_counters(
     google_id = await get_setting_value(db, GOOGLE_ADS_ID_KEY) or ''
     google_label = await get_setting_value(db, GOOGLE_ADS_LABEL_KEY) or ''
 
+    # Yandex Metrika offline conversions snapshot from Settings
+    oc_enabled = bool(getattr(settings, 'YANDEX_OFFLINE_CONV_ENABLED', False))
+    oc_counter = str(getattr(settings, 'YANDEX_OFFLINE_CONV_COUNTER_ID', '') or '')
+    oc_secret = str(getattr(settings, 'YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET', '') or '')
+    oc_secret_masked = ('*' * 8 + oc_secret[-4:]) if len(oc_secret) > 4 else ('***' if oc_secret else '')
+    oc_goals: list[OfflineConvGoal] = []
+    if oc_enabled:
+        oc_goals = [
+            OfflineConvGoal(name='Registration', event_id='registration', dedup='user_id'),
+            OfflineConvGoal(name='Trial', event_id='trial-add', dedup='user_id'),
+            OfflineConvGoal(name='Purchase', event_id='purchase', dedup='order_id'),
+        ]
+
     return AnalyticsCountersResponse(
         yandex_metrika_id=yandex_id,
         google_ads_id=google_id,
         google_ads_label=google_label,
+        offline_conv_enabled=oc_enabled,
+        offline_conv_counter_id=oc_counter,
+        offline_conv_measurement_secret_masked=oc_secret_masked,
+        offline_conv_goals=oc_goals,
     )
 
 
@@ -966,13 +995,56 @@ async def update_analytics_counters(
     google_id = await get_setting_value(db, GOOGLE_ADS_ID_KEY) or ''
     google_label = await get_setting_value(db, GOOGLE_ADS_LABEL_KEY) or ''
 
+    oc_enabled = bool(getattr(settings, 'YANDEX_OFFLINE_CONV_ENABLED', False))
+    oc_counter = str(getattr(settings, 'YANDEX_OFFLINE_CONV_COUNTER_ID', '') or '')
+    oc_secret = str(getattr(settings, 'YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET', '') or '')
+    oc_secret_masked = ('*' * 8 + oc_secret[-4:]) if len(oc_secret) > 4 else ('***' if oc_secret else '')
+    oc_goals: list[OfflineConvGoal] = []
+    if oc_enabled:
+        oc_goals = [
+            OfflineConvGoal(name='Registration', event_id='registration', dedup='user_id'),
+            OfflineConvGoal(name='Trial', event_id='trial-add', dedup='user_id'),
+            OfflineConvGoal(name='Purchase', event_id='purchase', dedup='order_id'),
+        ]
+
     return AnalyticsCountersResponse(
         yandex_metrika_id=yandex_id,
         google_ads_id=google_id,
         google_ads_label=google_label,
+        offline_conv_enabled=oc_enabled,
+        offline_conv_counter_id=oc_counter,
+        offline_conv_measurement_secret_masked=oc_secret_masked,
+        offline_conv_goals=oc_goals,
     )
 
 
+# ============ Yandex CID Sync ============
+
+
+class YandexCidRequest(BaseModel):
+    cid: str = Field(max_length=128, pattern=r'^[A-Za-z0-9._:-]{4,128}$')
+
+
+@router.post('/analytics/yandex-cid', status_code=204)
+async def store_yandex_cid(
+    body: YandexCidRequest,
+    user: User = Depends(get_current_cabinet_user),
+    db: AsyncSession = Depends(get_cabinet_db),
+):
+    """Store Yandex Metrika ClientID for the authenticated cabinet user."""
+    try:
+        from app.services import yandex_offline_conv_service as yandex_conv
+
+        await yandex_conv.store_cid(db, user.id, body.cid, source='cabinet')
+        await db.commit()
+    except Exception as exc:
+        logger.warning('Failed to store yandex_cid', user_id=user.id, exc=str(exc))
+        try:
+            await db.rollback()
+        except Exception:
+            pass
+
+
 # ============ Lite Mode Routes ============
 
 

app/config.py

@@ -567,6 +567,21 @@ class Settings(BaseSettings):
     KASSA_AI_SBERPAY_ENABLED: bool = False  # SberPay — payment_system_id=43
     KASSA_AI_SBERPAY_DISPLAY_NAME: str = 'SberPay (KassaAI)'
 
+    # ── Yandex Metrika offline conversions (server → mc.yandex.ru/collect) ──
+    YANDEX_OFFLINE_CONV_ENABLED: bool = False
+    YANDEX_OFFLINE_CONV_COUNTER_ID: str = ''
+    YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET: str = ''
+    YANDEX_OFFLINE_CONV_START_PREFIX: str = 'utm_ya_'
+    YANDEX_OFFLINE_CONV_DL: str = ''
+    YANDEX_OFFLINE_CONV_DT: str = ''
+    YANDEX_OFFLINE_CONV_CURRENCY: str = 'RUB'
+
+    # ── S2S Postback (server-to-server affiliate notifications) ──
+    S2S_POSTBACK_ENABLED: bool = False
+    S2S_POSTBACK_REGISTRATION_URL: str = ''
+    S2S_POSTBACK_TRIAL_URL: str = ''
+    S2S_POSTBACK_PURCHASE_URL: str = ''
+
     # RioPay (api.riopay.online) v2.0.1
     RIOPAY_ENABLED: bool = False
     RIOPAY_API_TOKEN: str | None = None  # x-api-token header

app/database/crud/yandex_client_id.py

@@ -0,0 +1,108 @@
+"""CRUD operations for yandex_client_id_map table."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import structlog
+from sqlalchemy import select, update
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database.models import YandexClientIdMap
+
+
+logger = structlog.get_logger(__name__)
+
+
+async def upsert_cid(
+    db: AsyncSession,
+    user_id: int,
+    cid: str,
+    source: str = 'web',
+    counter_id: str | None = None,
+    subid: str | None = None,
+) -> YandexClientIdMap:
+    """Insert or update Yandex ClientID for a user (race-safe via ON CONFLICT)."""
+    now = datetime.now(UTC)
+    values = {
+        'yandex_cid': cid,
+        'source': source,
+        'updated_at': now,
+    }
+    if counter_id:
+        values['counter_id'] = counter_id
+    if subid:
+        values['subid'] = subid
+
+    stmt = (
+        pg_insert(YandexClientIdMap)
+        .values(user_id=user_id, yandex_cid=cid, source=source, counter_id=counter_id, subid=subid)
+        .on_conflict_do_update(index_elements=['user_id'], set_=values)
+        .returning(YandexClientIdMap)
+    )
+
+    result = await db.execute(stmt)
+    await db.flush()
+    return result.scalar_one()
+
+
+async def get_cid(db: AsyncSession, user_id: int) -> YandexClientIdMap | None:
+    """Get Yandex ClientID mapping for a user."""
+    result = await db.execute(select(YandexClientIdMap).where(YandexClientIdMap.user_id == user_id))
+    return result.scalar_one_or_none()
+
+
+async def mark_registration_sent(db: AsyncSession, user_id: int) -> None:
+    """Mark registration event as sent for a user."""
+    await db.execute(
+        update(YandexClientIdMap)
+        .where(YandexClientIdMap.user_id == user_id)
+        .values(registration_sent=True, updated_at=datetime.now(UTC))
+    )
+    await db.flush()
+
+
+async def mark_trial_sent(db: AsyncSession, user_id: int) -> None:
+    """Mark trial event as sent for a user."""
+    await db.execute(
+        update(YandexClientIdMap)
+        .where(YandexClientIdMap.user_id == user_id)
+        .values(trial_sent=True, updated_at=datetime.now(UTC))
+    )
+    await db.flush()
+
+
+async def upsert_subid(
+    db: AsyncSession,
+    user_id: int,
+    subid: str,
+    source: str = 'web',
+) -> None:
+    """Save subid for a user. Updates existing record or creates with placeholder CID."""
+    if not subid or len(subid) > 255:
+        return
+    now = datetime.now(UTC)
+    # Try update first (don't create empty CID records)
+    result = await db.execute(
+        update(YandexClientIdMap).where(YandexClientIdMap.user_id == user_id).values(subid=subid, updated_at=now)
+    )
+    if result.rowcount == 0:
+        # No existing record — create with placeholder
+        stmt = (
+            pg_insert(YandexClientIdMap)
+            .values(user_id=user_id, yandex_cid='_subid_only', source=source, subid=subid)
+            .on_conflict_do_update(
+                index_elements=['user_id'],
+                set_={'subid': subid, 'updated_at': now},
+            )
+        )
+        await db.execute(stmt)
+    await db.flush()
+    logger.info('Subid saved', user_id=user_id, subid=subid, source=source)
+
+
+async def get_subid(db: AsyncSession, user_id: int) -> str | None:
+    """Get subid for a user."""
+    result = await db.execute(select(YandexClientIdMap.subid).where(YandexClientIdMap.user_id == user_id))
+    return result.scalar_one_or_none()

app/database/models.py

@@ -3528,6 +3528,10 @@ class GuestPurchase(Base):
     retry_count = Column(Integer, nullable=False, default=0, server_default='0')
     receipt_uuid = Column(String(255), nullable=True, index=True)
     receipt_created_at = Column(AwareDateTime(), nullable=True)
+    # Yandex Metrika offline conversions: client identifier + traffic source tags
+    yandex_cid = Column(String(128), nullable=True)
+    subid = Column(String(255), nullable=True)
+    referrer = Column(String(500), nullable=True)
 
     landing = relationship('LandingPage', back_populates='guest_purchases', lazy='selectin')
     tariff = relationship('Tariff', lazy='selectin')
@@ -3609,3 +3613,26 @@ class NewsTag(Base):
 
     def __repr__(self) -> str:
         return f"<NewsTag id={self.id} name='{self.name}'>"
+
+
+class YandexClientIdMap(Base):
+    """Yandex Metrika client identifier captured per user.
+
+    Stores the mapping user_id -> yandex_cid so we can fire offline
+    conversion events to mc.yandex.ru with the right CID even after
+    the user leaves the landing/web flow. The ``subid`` column carries
+    a pass-through traffic-source identifier for S2S postbacks.
+    """
+
+    __tablename__ = 'yandex_client_id_map'
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'), unique=True, nullable=False)
+    yandex_cid = Column(String(128), nullable=False)
+    source = Column(String(20), nullable=False, default='web', server_default='web')
+    counter_id = Column(String(32), nullable=True)
+    registration_sent = Column(Boolean, default=False, server_default=text('false'), nullable=False)
+    trial_sent = Column(Boolean, default=False, server_default=text('false'), nullable=False)
+    subid = Column(String(255), nullable=True)
+    created_at = Column(AwareDateTime(), server_default=func.now())
+    updated_at = Column(AwareDateTime(), server_default=func.now(), onupdate=func.now())

app/services/s2s_postback_service.py

@@ -0,0 +1,92 @@
+"""S2S Postback Service — sends server-to-server postbacks on events."""
+
+import structlog
+
+from app.config import settings
+
+
+logger = structlog.get_logger(__name__)
+
+try:
+    import httpx
+except ImportError:
+    httpx = None
+
+
+def _is_enabled() -> bool:
+    return getattr(settings, 'S2S_POSTBACK_ENABLED', False) and httpx is not None
+
+
+def _get_url(event: str) -> str | None:
+    """Get postback URL template for event type."""
+    mapping = {
+        'registration': getattr(settings, 'S2S_POSTBACK_REGISTRATION_URL', ''),
+        'trial': getattr(settings, 'S2S_POSTBACK_TRIAL_URL', ''),
+        'purchase': getattr(settings, 'S2S_POSTBACK_PURCHASE_URL', ''),
+    }
+    url = mapping.get(event, '')
+    return url or None
+
+
+async def send_postback(
+    event: str,
+    subid: str,
+    amount: float | None = None,
+    user_id: int | None = None,
+) -> bool:
+    """Send S2S postback for an event.
+
+    Args:
+        event: 'registration', 'trial', or 'purchase'
+        subid: tracking subid from URL
+        amount: purchase amount in rubles (for purchase event)
+        user_id: internal user ID for logging
+
+    Returns:
+        True if sent successfully
+    """
+    if not _is_enabled():
+        return False
+
+    if not subid:
+        return False
+
+    url_template = _get_url(event)
+    if not url_template:
+        logger.debug('S2S postback URL not configured', event=event)
+        return False
+
+    # Replace placeholders (URL-encode subid to prevent injection)
+    from urllib.parse import quote
+
+    url = url_template.replace('{subid}', quote(subid, safe=''))
+    url = url.replace('{event}', event)
+    if amount is not None:
+        url = url.replace('{amount}', str(round(amount, 2)))
+    else:
+        url = url.replace('{amount}', '0')
+
+    url = url.replace('{user_id}', str(user_id) if user_id is not None else '0')
+
+    try:
+        async with httpx.AsyncClient(timeout=10) as client:
+            response = await client.get(url)
+            logger.info(
+                'S2S postback sent',
+                event=event,
+                subid=subid,
+                amount=amount,
+                user_id=user_id,
+                status_code=response.status_code,
+                url=url[:100],
+            )
+            return response.status_code < 400
+    except Exception as e:
+        logger.error(
+            'S2S postback failed',
+            event=event,
+            subid=subid,
+            error=str(e),
+            url=url[:100],
+        )
+        return False

app/services/yandex_offline_conv_service.py

@@ -0,0 +1,352 @@
+"""Yandex.Metrika offline conversions service.
+
+Sends events (registration, trial-add, purchase) to mc.yandex.ru/collect
+using the Measurement Protocol. No pageview needed — user has active
+Metrika session from the site. yclid is passed via landing page URL,
+Metrika matches it automatically.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+import time
+
+import httpx
+import structlog
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.database.crud.yandex_client_id import (
+    get_cid,
+    mark_registration_sent,
+    mark_trial_sent,
+    upsert_cid,
+)
+from app.database.database import AsyncSessionLocal
+
+
+logger = structlog.get_logger(__name__)
+
+COLLECT_URL = 'https://mc.yandex.ru/collect'
+TIMEOUT = 10.0
+MAX_RETRIES = 3
+RETRY_DELAY = 1.0
+
+_CID_RE = re.compile(r'^[A-Za-z0-9._:-]{4,128}$')
+_http_client: httpx.AsyncClient | None = None
+
+
+def _get_client() -> httpx.AsyncClient:
+    global _http_client
+    if _http_client is None or _http_client.is_closed:
+        _http_client = httpx.AsyncClient(timeout=TIMEOUT)
+    return _http_client
+
+
+def _is_enabled() -> bool:
+    return bool(
+        settings.YANDEX_OFFLINE_CONV_ENABLED
+        and settings.YANDEX_OFFLINE_CONV_COUNTER_ID
+        and settings.YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET
+    )
+
+
+def _normalize_cid(cid: str | None) -> str | None:
+    if not isinstance(cid, str):
+        return None
+    cid = cid.strip()
+    if not cid or not _CID_RE.match(cid):
+        return None
+    return cid
+
+
+def _mask_cid(cid: str) -> str:
+    if len(cid) <= 4:
+        return '****'
+    return '*' * (len(cid) - 4) + cid[-4:]
+
+
+def _base_payload(cid: str) -> dict[str, str]:
+    return {
+        'tid': settings.YANDEX_OFFLINE_CONV_COUNTER_ID,
+        'cid': cid,
+        'ms': settings.YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET,
+    }
+
+
+def _pageview_payload(cid: str) -> dict[str, str]:
+    payload = _base_payload(cid)
+    payload.update(
+        {
+            't': 'pageview',
+            'dl': settings.YANDEX_OFFLINE_CONV_DL or 'https://web.mtrxvps.ru',
+            'dt': settings.YANDEX_OFFLINE_CONV_DT or 'Matrixxx VPN',
+        }
+    )
+    return payload
+
+
+def _event_payload(cid: str, event_action: str) -> dict[str, str]:
+    payload = _base_payload(cid)
+    payload.update(
+        {
+            't': 'event',
+            'ea': event_action,
+        }
+    )
+    return payload
+
+
+def _ecommerce_purchase_payload(
+    cid: str,
+    amount_rubles: float,
+    order_id: str = '',
+    product_name: str = '',
+    product_category: str = '',
+) -> dict[str, str]:
+    """Build ecommerce:purchase payload for Metrika Measurement Protocol."""
+
+    service_name = (
+        getattr(settings, 'YANDEX_OFFLINE_CONV_DT', '')
+        or getattr(settings, 'PAYMENT_SERVICE_NAME', '')
+        or 'Subscription'
+    )
+    currency = getattr(settings, 'YANDEX_OFFLINE_CONV_CURRENCY', '') or 'RUB'
+    payload = _base_payload(cid)
+    payload.update(
+        {
+            't': 'event',
+            'ea': 'purchase',
+            'pa': 'purchase',
+            'ti': order_id or str(int(time.time())),
+            'tr': str(amount_rubles),
+            'cu': currency,
+            'ev': str(amount_rubles),
+            'pr1id': 'subscription',
+            'pr1nm': product_name or service_name,
+            'pr1ca': product_category or 'subscription',
+            'pr1pr': str(amount_rubles),
+            'pr1qt': '1',
+        }
+    )
+    return payload
+
+
+async def _post_collect(payload: dict[str, str], kind: str, cid: str) -> bool:
+    """POST to mc.yandex.ru/collect with retries. Returns True on success."""
+    masked = _mask_cid(cid)
+    for attempt in range(1, MAX_RETRIES + 1):
+        try:
+            client = _get_client()
+            resp = await client.post(COLLECT_URL, data=payload)
+
+            if 200 <= resp.status_code < 300:
+                logger.info('collect sent', kind=kind, cid=masked, status=resp.status_code)
+                return True
+
+            if 500 <= resp.status_code < 600 and attempt < MAX_RETRIES:
+                logger.warning(
+                    'collect server error',
+                    kind=kind,
+                    attempt=attempt,
+                    max=MAX_RETRIES,
+                    cid=masked,
+                    status=resp.status_code,
+                )
+                await asyncio.sleep(RETRY_DELAY)
+                continue
+
+            logger.error('collect rejected', kind=kind, cid=masked, status=resp.status_code, body=resp.text[:200])
+            return False
+
+        except Exception as exc:
+            logger.warning(
+                'collect request error', kind=kind, attempt=attempt, max=MAX_RETRIES, cid=masked, error=str(exc)
+            )
+            if attempt < MAX_RETRIES:
+                await asyncio.sleep(RETRY_DELAY)
+                continue
+            return False
+
+    return False
+
+
+async def _send_event(cid: str, event_action: str) -> bool:
+    """Send event directly — no pageview needed, user has active Metrika session."""
+    return await _post_collect(_event_payload(cid, event_action), event_action, cid)
+
+
+# --- Background task helpers ---
+
+_background_tasks: set[asyncio.Task] = set()
+
+
+def _task_done(task):
+    """Log errors from background conversion tasks."""
+    _background_tasks.discard(task)
+    if task.cancelled():
+        return
+    exc = task.exception()
+    if exc:
+        logger.error('YandexOfflineConv background task failed', error=str(exc))
+
+
+def spawn_bg(coro) -> None:
+    """Spawn a background Yandex conversion task with proper reference tracking.
+
+    Checks _is_enabled() early so callers don't need to.
+    """
+    if not _is_enabled():
+        # Close the coroutine to avoid RuntimeWarning
+        coro.close()
+        return
+    task = asyncio.create_task(coro)
+    _background_tasks.add(task)
+    task.add_done_callback(_task_done)
+
+
+async def _fire_bg(event_name: str, event_fn, user_id: int, **kwargs) -> None:
+    """Generic background wrapper: opens a session, calls event_fn, logs errors."""
+    try:
+        async with AsyncSessionLocal() as db:
+            await event_fn(db, user_id, **kwargs)
+    except Exception as exc:
+        logger.warning('YandexOfflineConv background event failed', event=event_name, user_id=user_id, error=str(exc))
+
+
+async def fire_registration_bg(user_id: int) -> None:
+    """Fire registration event in background with its own DB session."""
+    await _fire_bg('registration', on_registration, user_id)
+
+
+async def fire_trial_bg(user_id: int) -> None:
+    """Fire trial event in background with its own DB session."""
+    await _fire_bg('trial', on_trial, user_id)
+
+
+async def fire_purchase_bg(user_id: int, amount_kopeks: int) -> None:
+    """Fire purchase event in background with its own DB session."""
+    await _fire_bg('purchase', on_purchase, user_id, amount_kopeks=amount_kopeks)
+
+
+# --- Public API ---
+async def store_cid(
+    db: AsyncSession,
+    user_id: int,
+    cid: str | None,
+    source: str = 'web',
+) -> bool:
+    """Store Yandex ClientID for a user. Returns True if stored."""
+    normalized = _normalize_cid(cid)
+    if not normalized:
+        return False
+
+    try:
+        await upsert_cid(db, user_id, normalized, source=source, counter_id=settings.YANDEX_OFFLINE_CONV_COUNTER_ID)
+        logger.info('stored CID', user_id=user_id, source=source)
+        return True
+    except Exception as exc:
+        logger.error('failed to store CID', user_id=user_id, error=str(exc))
+        return False
+
+
+async def store_cid_and_fire_registration(
+    user_id: int,
+    cid: str | None,
+    *,
+    source: str = 'web',
+) -> None:
+    """Store Yandex CID and fire registration conversion in background (best-effort).
+
+    Opens its own DB session so it never interferes with the caller's transaction.
+    """
+    if not cid:
+        return
+    try:
+        async with AsyncSessionLocal() as db:
+            stored = await store_cid(db, user_id, cid, source=source)
+            if stored:
+                await db.commit()
+                spawn_bg(fire_registration_bg(user_id))
+    except Exception as exc:
+        logger.warning('Failed to store CID and fire registration', user_id=user_id, error=str(exc))
+
+
+async def on_registration(db: AsyncSession, user_id: int) -> None:
+    """Fire registration event (once per user)."""
+    if not _is_enabled():
+        return
+
+    try:
+        row = await get_cid(db, user_id)
+        if not row or row.registration_sent:
+            return
+        if not row.yandex_cid or row.yandex_cid.startswith('_'):
+            return  # placeholder row — real CID not yet received
+
+        success = await _send_event(row.yandex_cid, 'registration')
+        if success:
+            await mark_registration_sent(db, user_id)
+            await db.commit()
+            logger.info('registration event sent', user_id=user_id)
+    except Exception as exc:
+        logger.error('registration event failed', user_id=user_id, error=str(exc))
+
+
+async def on_trial(db: AsyncSession, user_id: int) -> None:
+    """Fire trial-add event (once per user)."""
+    if not _is_enabled():
+        return
+
+    try:
+        row = await get_cid(db, user_id)
+        if not row or row.trial_sent:
+            return
+        if not row.yandex_cid or row.yandex_cid.startswith('_'):
+            return  # placeholder row — real CID not yet received
+
+        success = await _send_event(row.yandex_cid, 'trial-add')
+        if success:
+            await mark_trial_sent(db, user_id)
+            await db.commit()
+            logger.info('trial-add event sent', user_id=user_id)
+    except Exception as exc:
+        logger.error('trial-add event failed', user_id=user_id, error=str(exc))
+
+
+async def on_purchase(db: AsyncSession, user_id: int, amount_kopeks: int) -> None:
+    """Fire ecommerce purchase event (every payment)."""
+    if not _is_enabled():
+        return
+
+    try:
+        row = await get_cid(db, user_id)
+        if not row:
+            return
+        if not row.yandex_cid or row.yandex_cid.startswith('_'):
+            return  # placeholder row — real CID not yet received
+
+        amount_rubles = amount_kopeks / 100
+        payload = _ecommerce_purchase_payload(row.yandex_cid, amount_rubles)
+        success = await _post_collect(payload, 'purchase', row.yandex_cid)
+        if success:
+            logger.info('purchase event sent', user_id=user_id, amount=amount_rubles)
+    except Exception as exc:
+        logger.error('purchase event failed', user_id=user_id, error=str(exc))
+
+
+def parse_cid_from_start_param(param: str) -> tuple[str | None, str]:
+    """Extract Yandex CID from bot start parameter.
+
+    If param starts with the configured prefix (e.g. 'utm_ya_'),
+    returns (cid, original_param). Otherwise returns (None, original_param).
+    Original param is always preserved for UTM tracking.
+    """
+    prefix = settings.YANDEX_OFFLINE_CONV_START_PREFIX
+    if not prefix or not param.startswith(prefix):
+        return None, param
+
+    cid = param[len(prefix) :]
+    normalized = _normalize_cid(cid)
+    return normalized, param  # Keep original param for UTM tracking

migrations/alembic/versions/0063_add_yandex_client_id_map.py

@@ -0,0 +1,99 @@
+"""add yandex_client_id_map table + guest_purchases offline conv columns
+
+Revision ID: 0063
+Revises: 0062
+Create Date: 2026-04-21
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = '0063'
+down_revision: Union[str, None] = '0062'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    # 1) yandex_client_id_map — created idempotently
+    result = conn.execute(
+        sa.text("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'yandex_client_id_map')")
+    )
+    if not result.scalar():
+        op.create_table(
+            'yandex_client_id_map',
+            sa.Column('id', sa.Integer, primary_key=True, autoincrement=True),
+            sa.Column(
+                'user_id',
+                sa.Integer,
+                sa.ForeignKey('users.id', ondelete='CASCADE'),
+                unique=True,
+                nullable=False,
+            ),
+            sa.Column('yandex_cid', sa.String(128), nullable=False),
+            sa.Column('source', sa.String(20), nullable=False, server_default='web'),
+            sa.Column('counter_id', sa.String(32), nullable=True),
+            sa.Column(
+                'registration_sent',
+                sa.Boolean,
+                nullable=False,
+                server_default=sa.text('false'),
+            ),
+            sa.Column(
+                'trial_sent',
+                sa.Boolean,
+                nullable=False,
+                server_default=sa.text('false'),
+            ),
+            sa.Column('subid', sa.String(255), nullable=True),
+            sa.Column(
+                'created_at',
+                sa.DateTime(timezone=True),
+                server_default=sa.func.now(),
+            ),
+            sa.Column(
+                'updated_at',
+                sa.DateTime(timezone=True),
+                server_default=sa.func.now(),
+            ),
+        )
+
+    # 2) guest_purchases — add yandex_cid / subid / referrer (idempotent)
+    for col_name, col_def in (
+        ('yandex_cid', sa.Column('yandex_cid', sa.String(128), nullable=True)),
+        ('subid', sa.Column('subid', sa.String(255), nullable=True)),
+        ('referrer', sa.Column('referrer', sa.String(500), nullable=True)),
+    ):
+        result = conn.execute(
+            sa.text(
+                'SELECT EXISTS (SELECT 1 FROM information_schema.columns '
+                "WHERE table_name = 'guest_purchases' AND column_name = :col)"
+            ),
+            {'col': col_name},
+        )
+        if not result.scalar():
+            op.add_column('guest_purchases', col_def)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    for col_name in ('referrer', 'subid', 'yandex_cid'):
+        result = conn.execute(
+            sa.text(
+                'SELECT EXISTS (SELECT 1 FROM information_schema.columns '
+                "WHERE table_name = 'guest_purchases' AND column_name = :col)"
+            ),
+            {'col': col_name},
+        )
+        if result.scalar():
+            op.drop_column('guest_purchases', col_name)
+
+    result = conn.execute(
+        sa.text("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'yandex_client_id_map')")
+    )
+    if result.scalar():
+        op.drop_table('yandex_client_id_map')