
Клиентские пиксели и JS-SDK теряют до 20–60% событий из‑за:
Надёжный путь — проксировать события через собственный домен и отправлять их на сервер, где уже устойчиво догружать в внешние системы: GA4 (Measurement Protocol) и Meta CAPI. Мы контролируем хранение click-id (gclid, fbclid), работаем с первопартийными cookies, управляем дедупликацией и соблюдаем согласия. Ниже — практический план с рабочими примерами кода.
Итог: стабильная атрибуция, меньше утечек данных и гибкое соблюдение приватности.
Пример для Next.js middleware (работает на edge, не зависит от рендера конкретной страницы):
// middleware.ts
import { NextResponse } from 'next/server'
import type { NextRequest } from 'next/server'
function setCookie(res: NextResponse, name: string, value: string, maxAgeDays = 90) {
const maxAge = maxAgeDays * 24 * 60 * 60
res.cookies.set(name, value, {
httpOnly: false,
secure: true,
sameSite: 'Lax',
path: '/',
maxAge,
})
}
function genFbp(ts: number) {
const rnd = Math.floor(Math.random() * 1e10)
return `fb.1.${ts}.${rnd}`
}
export function middleware(req: NextRequest) {
const res = NextResponse.next()
const url = req.nextUrl
const p = url.searchParams
const ts = Date.now()
const gclid = p.get('gclid') || p.get('gbraid') || p.get('wbraid')
const fbclid = p.get('fbclid')
// UTM-метки
const utmKeys = ['utm_source','utm_medium','utm_campaign','utm_content','utm_term']
utmKeys.forEach(k => {
const v = p.get(k)
if (v) setCookie(res, k, v)
})
// GA client id (наша первопартийная альтернатива). Если нет — создадим.
const cid = req.cookies.get('cid')?.value
if (!cid) {
const newCid = crypto.randomUUID()
setCookie(res, 'cid', newCid, 365)
}
// Click IDs: Google
if (gclid) setCookie(res, 'gclid', gclid)
// Meta: fbp/fbc
const fbp = req.cookies.get('fbp')?.value || genFbp(ts)
if (!req.cookies.get('fbp')) setCookie(res, 'fbp', fbp)
if (fbclid) {
const fbc = `fb.1.${Math.floor(ts/1000)}.${fbclid}`
setCookie(res, 'fbc', fbc)
}
return res
}
export const config = {
matcher: ['/((?!_next/static|_next/image|favicon.ico).*)'],
}
Это даёт нам первопартийные идентификаторы: cid, fbp/fbc, utm_*. Они пригодятся на сервере.
На фронте отправляем события на свой сервер, а не в сторонние SDK.
// lib/track.ts
export async function track(event: string, payload: Record<string, any> = {}) {
const body = {
event, // например 'page_view', 'purchase', 'lead'
payload,
ts: Date.now(),
}
// Можно подписывать событие HMAC, чтобы не принимать мусор снаружи
const secret = process.env.NEXT_PUBLIC_TRACKING_HMAC_SECRET || ''
const enc = new TextEncoder()
const key = await crypto.subtle.importKey(
'raw', enc.encode(secret), { name: 'HMAC', hash: 'SHA-256' }, false, ['sign']
)
const signatureBuf = await crypto.subtle.sign('HMAC', key, enc.encode(JSON.stringify(body)))
const signature = Array.from(new Uint8Array(signatureBuf)).map(b => b.toString(16).padStart(2, '0')).join('')
await fetch('/e', {
method: 'POST',
headers: {
'content-type': 'application/json',
'x-signature': signature,
},
body: JSON.stringify(body),
keepalive: true, // чтобы отправлялось при закрытии страницы
credentials: 'include',
})
}
Использование:
import { track } from './lib/track'
track('page_view', { title: document.title, path: location.pathname })
// Пример покупки
track('purchase', {
value: 1990,
currency: 'RUB',
order_id: 'ORD-1001',
items: [
{ item_id: 'sku-1', item_name: 'Куртка', price: 1990, quantity: 1 },
],
// Опционально передаём email/phone для CAPI (будет хеширован на сервере)
email: 'user@example.com',
phone: '+79991234567',
})
Ниже минимально жизнеспособная реализация, готовая к запуску. Для продакшена добавьте Redis вместо in-memory очереди и TTLCache, секреты — через env, ограничения по скорости и DLQ.
# app.py
import asyncio
import hashlib
import hmac
import json
import os
import time
from typing import Any, Dict, Optional
import httpx
from fastapi import FastAPI, Request, Header, HTTPException
from pydantic import BaseModel, Field
from cachetools import TTLCache
app = FastAPI()
# Конфигурация через ENV
GA4_MEASUREMENT_ID = os.getenv('GA4_MEASUREMENT_ID', 'G-XXXXXXX')
GA4_API_SECRET = os.getenv('GA4_API_SECRET', 'secret')
GA4_ENDPOINT = 'https://www.google-analytics.com/mp/collect'
META_PIXEL_ID = os.getenv('META_PIXEL_ID', '1234567890')
META_ACCESS_TOKEN = os.getenv('META_ACCESS_TOKEN', 'EAAXXXXX')
META_ENDPOINT = f'https://graph.facebook.com/v18.0/{META_PIXEL_ID}/events'
META_TEST_EVENT_CODE = os.getenv('META_TEST_EVENT_CODE') # опционально для тестов
TRACKING_HMAC_SECRET = os.getenv('TRACKING_HMAC_SECRET', 'dev-secret')
SEND_TIMEOUT = 8.0
MAX_RETRIES = 5
# Очередь и дедуп
queue: 'asyncio.Queue[dict]' = asyncio.Queue(maxsize=10000)
dedup_cache = TTLCache(maxsize=100000, ttl=48*3600) # 48 часов
client = httpx.AsyncClient(timeout=SEND_TIMEOUT)
class EventIn(BaseModel):
event: str = Field(..., min_length=2, max_length=64)
payload: Dict[str, Any] = {}
ts: int = Field(default_factory=lambda: int(time.time()*1000))
class EnrichedEvent(BaseModel):
event: str
payload: Dict[str, Any]
ts: int
ip: Optional[str]
ua: Optional[str]
cid: Optional[str]
gclid: Optional[str]
fbp: Optional[str]
fbc: Optional[str]
utm: Dict[str, Optional[str]] = {}
event_id: str # единый id для дедупа Meta
# Утилиты
def sha256_hex(v: str) -> str:
return hashlib.sha256(v.encode('utf-8')).hexdigest()
async def backoff_attempts():
for i in range(MAX_RETRIES):
delay = min(0.25 * (2 ** i), 20)
yield i, delay
# GA4 отправка
async def send_ga4(ev: EnrichedEvent):
params = {
'measurement_id': GA4_MEASUREMENT_ID,
'api_secret': GA4_API_SECRET,
}
client_id = ev.cid or f'anon.{sha256_hex((ev.ip or "0") + (ev.ua or "0"))[:16]}'
ga4_event = {
'client_id': client_id,
'timestamp_micros': ev.ts * 1000,
'events': [
{
'name': ev.event,
'params': {
**{k: v for k, v in ev.payload.items() if k not in ['email','phone','external_id']},
'event_id': ev.event_id,
'engagement_time_msec': 1,
}
}
],
}
# Consent можно учитывать так (если храните флаг в cookies):
# ga4_event['consent'] = {'ad_user_data': 'N', 'ad_personalization': 'N'}
for attempt, delay in [*[(0, 0)], *[(i+1, d) for i, d in await backoff_attempts().__anext__()]]:
try:
r = await client.post(GA4_ENDPOINT, params=params, json=ga4_event)
if r.status_code < 300:
return
if r.status_code in (429, 500, 502, 503, 504):
await asyncio.sleep(delay)
continue
raise RuntimeError(f'GA4 error {r.status_code}: {r.text}')
except StopAsyncIteration:
break
# Meta CAPI отправка
async def send_meta(ev: EnrichedEvent, test_mode: bool = False):
user_data: Dict[str, Any] = {}
# Хеширование PII согласно требованиям Meta
email = ev.payload.get('email')
if email:
email_clean = email.strip().lower()
user_data['em'] = sha256_hex(email_clean)
phone = ev.payload.get('phone')
if phone:
digits = ''.join([c for c in phone if c.isdigit()])
user_data['ph'] = sha256_hex(digits)
if ev.ip:
user_data['client_ip_address'] = ev.ip
if ev.ua:
user_data['client_user_agent'] = ev.ua
if ev.fbp:
user_data['fbp'] = ev.fbp
if ev.fbc:
user_data['fbc'] = ev.fbc
action_source = 'website'
event_source_url = ev.payload.get('url')
custom_data: Dict[str, Any] = {}
for k in ['value','currency','order_id']:
if k in ev.payload:
custom_data[k] = ev.payload[k]
if 'items' in ev.payload:
custom_data['contents'] = [
{
'id': it.get('item_id') or it.get('id'),
'quantity': it.get('quantity', 1),
'item_price': it.get('price'),
} for it in ev.payload['items']
]
custom_data['content_type'] = 'product'
meta_event = {
'data': [
{
'event_name': ev.event,
'event_time': int(ev.ts/1000),
'event_id': ev.event_id,
'action_source': action_source,
**({'event_source_url': event_source_url} if event_source_url else {}),
'user_data': user_data,
**({'custom_data': custom_data} if custom_data else {}),
}
],
'access_token': META_ACCESS_TOKEN,
}
if test_mode and META_TEST_EVENT_CODE:
meta_event['test_event_code'] = META_TEST_EVENT_CODE
for i in range(MAX_RETRIES):
r = await client.post(META_ENDPOINT, json=meta_event)
if r.status_code < 300:
return
if r.status_code in (429, 500, 502, 503, 504):
await asyncio.sleep(min(0.25*(2**i), 20))
continue
raise RuntimeError(f'Meta error {r.status_code}: {r.text}')
# Подпись запроса от фронта (HMAC)
def verify_signature(body_bytes: bytes, signature_hex: str) -> bool:
mac = hmac.new(TRACKING_HMAC_SECRET.encode('utf-8'), body_bytes, hashlib.sha256).hexdigest()
return hmac.compare_digest(mac, signature_hex)
# Нормализация входящего события и постановка в очередь
@app.post('/e')
async def ingest(request: Request, x_signature: Optional[str] = Header(default=None)):
body_bytes = await request.body()
try:
data = json.loads(body_bytes)
except Exception:
raise HTTPException(400, 'invalid json')
if not x_signature or not verify_signature(body_bytes, x_signature):
raise HTTPException(401, 'invalid signature')
ev = EventIn(**data)
# Consent. Если есть cookie consent_marketing=0 — игнорируем маркетинговые события
consent_cookie = request.cookies.get('consent_marketing')
if consent_cookie == '0':
return {'status': 'skipped: no consent'}
ip = request.headers.get('x-forwarded-for', request.client.host if request.client else None)
if ip and ',' in ip:
ip = ip.split(',')[0].strip()
ua = request.headers.get('user-agent')
# Извлечение первопартийных идентификаторов
cid = request.cookies.get('cid')
gclid = request.cookies.get('gclid')
fbp = request.cookies.get('fbp')
fbc = request.cookies.get('fbc')
utm = {k: request.cookies.get(k) for k in ['utm_source','utm_medium','utm_campaign','utm_content','utm_term']}
# Генерируем единый event_id для дедупликации между каналами
seed = f"{ev.event}|{ev.ts}|{cid}|{ev.payload.get('order_id') or ev.payload.get('lead_id') or ''}"
event_id = sha256_hex(seed)[:32]
enriched = EnrichedEvent(
event=ev.event,
payload={**ev.payload, **{'utm': utm, 'gclid': gclid}},
ts=ev.ts,
ip=ip,
ua=ua,
cid=cid,
gclid=gclid,
fbp=fbp,
fbc=fbc,
utm=utm,
event_id=event_id,
)
# Дедуп: если уже видели такой event_id — не дублируем
if event_id in dedup_cache:
return {'status': 'duplicate'}
dedup_cache[event_id] = True
await queue.put(enriched.dict())
return {'status': 'queued', 'event_id': event_id}
# Воркер отправки
async def worker():
while True:
item = await queue.get()
ev = EnrichedEvent(**item)
# Отправляем параллельно, но ошибки отдельно логируем
try:
await asyncio.gather(
send_ga4(ev),
send_meta(ev, test_mode=bool(META_TEST_EVENT_CODE)),
)
except Exception as e:
print('Dispatch error:', e)
finally:
queue.task_done()
@app.on_event('startup')
async def on_startup():
asyncio.create_task(worker())
@app.on_event('shutdown')
async def on_shutdown():
await client.aclose()
Запуск:
uvicorn app:app --host 0.0.0.0 --port 8000
Проверьте, что домен бэкенда совпадает с фронтом (или настроен CORS и cookies с SameSite=Lax/None). Лучше прокинуть /e через reverse-proxy на том же домене, чтобы события выглядели как «первопартийные» и не блокировались фильтрами.
Примеры curl для ручной проверки приёмника:
body='{ "event":"purchase", "payload": {"value": 990, "currency":"RUB", "order_id":"ORD-2002"}, "ts":'$(date +%s%3N)'}'
mac=$(echo -n $body | openssl dgst -sha256 -hmac "dev-secret" | cut -d' ' -f2)
curl -i -X POST http://localhost:8000/e \
-H "content-type: application/json" \
-H "x-signature: $mac" \
-H "user-agent: TestUA" \
-H "x-forwarded-for: 198.51.100.10" \
-b "cid=localcid123; fbp=fb.1.$(date +%s).123456; fbc=fb.1.$(date +%s).fbclid123; utm_source=ads; utm_medium=cpc; utm_campaign=brand" \
--data "$body"
Server-side трекинг не заменяет полностью клиентские пиксели, но делает систему устойчивой и управляемой. Начните с базового прокси, зафиксируйте click-id и client_id, обеспечьте дедупликацию и consent — и уже через неделю вы увидите, как отчёты и автостратегии становятся стабильнее.