feature: API-ключи клиентов и журнал использования в Postgres (#4, #17)

- Postgres в compose; пакет db (SQLAlchemy async): clients, windows, requests.
- Все эндпоинты, кроме /health, — только с ключом в X-API-Key, иначе 401; без
  выпущенных ключей доступа нет. В базе — только sha256 ключа.
- Задача и окно сессии привязаны к клиенту: чужой ключ их не видит.
- Воркер пишет журнал: окно (открыто, последнее обращение, закрыто) и задачу
  (в очереди, начало, конец, итог, данные без секретов). Сбой записи не роняет задачу.
- python -m db.cli create|list|revoke|stats; make key-create/key-list/key-revoke/stats.
- smoke и loadtest с ключом.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/api/logs.py

@@ -13,7 +13,8 @@ def mask_proxy(proxy_url: str | None) -> str | None:
 	return f"{parts.scheme}://{'***@' if parts.username else ''}{parts.hostname}:{parts.port}"
 
 
-def safe_request_dump(payload: SolveRequest) -> str:
+def safe_request_data(payload: SolveRequest) -> dict:
+	"""Запрос без секретов: для лога и журнала запросов (#17)."""
 	data = payload.model_dump(exclude_none=True)
 	data["proxy"] = mask_proxy(payload.proxy)
 	if payload.captcha:
@@ -23,7 +24,11 @@ def safe_request_dump(payload: SolveRequest) -> str:
 	for act in data["actions"]:
 		if act["type"] in ("fill", "typing") and "value" in act:
 			act["value"] = "***"
-	return json.dumps(data, indent=2, ensure_ascii=False)
+	return data
+
+
+def safe_request_dump(payload: SolveRequest) -> str:
+	return json.dumps(safe_request_data(payload), indent=2, ensure_ascii=False)
 
 
 def safe_response_dump(result: dict) -> str:

src/db/__init__.py

@@ -0,0 +1,6 @@
+"""
+Postgres (#4, #17): клиенты с API-ключами и журнал работы — окна браузера и запросы.
+
+Общий для api и worker. Схема создаётся при старте сервиса (`init_db`). `DATABASE_URL` —
+`postgresql+asyncpg://...`; в тестах — `sqlite+aiosqlite://...`.
+"""

src/db/cli.py

@@ -0,0 +1,67 @@
+"""
+Клиенты и API-ключи:
+
+	python -m db.cli create <name>   # выпустить ключ: печатается один раз
+	python -m db.cli list
+	python -m db.cli revoke <id>
+	python -m db.cli stats [--days 7]  # запросы, их время и время работы браузера по клиентам
+
+В compose — через make: make key-create name=..., make key-list, make key-revoke id=..., make stats days=...
+"""
+import argparse
+import asyncio
+
+from db import clients, stats
+from db.engine import connect
+
+
+async def run(args) -> None:
+	db = connect()
+	try:
+		await db.init()
+		if args.command == "create":
+			client, key = await clients.create_client(db, args.name)
+			print(f"client {client.id} ({client.name}), API key — сохраните, больше он не показывается:\n{key}")
+		elif args.command == "list":
+			for c in await clients.list_clients(db):
+				state = f"revoked {c.revoked_at:%Y-%m-%d %H:%M}" if c.revoked_at else "active"
+				last = f"{c.last_access_at:%Y-%m-%d %H:%M}" if c.last_access_at else "never"
+				print(f"{c.id:>4}  {c.key_prefix}…  {state:<24} last access {last:<16}  {c.name}")
+		elif args.command == "revoke":
+			print("revoked" if await clients.revoke(db, args.id) else "no active client with this id")
+		elif args.command == "stats":
+			print_stats(await stats.usage(db, stats.since_days(args.days)), args.days)
+	finally:
+		await db.close()
+
+
+def _ms(value: int | None) -> str:
+	return "—" if value is None else f"{value / 1000:.1f}s"
+
+
+def print_stats(rows: list, days: float) -> None:
+	print(f"за {days:g} дн.")
+	print(f"{'клиент':<24} {'запросов':>8} {'ok':>6} {'ср.':>7} {'p95':>7} {'окон':>5} {'браузер, ч':>10}  ошибки")
+	for u in rows:
+		errors = ", ".join(f"{k} {v}" for k, v in sorted(u.errors.items())) or "—"
+		name = f"{u.client_id or '-'} {u.name}"[:24]
+		print(
+			f"{name:<24} {u.requests:>8} {u.ok:>6} {_ms(u.avg_ms):>7} {_ms(u.p95_ms):>7} "
+			f"{u.windows:>5} {u.browser_seconds / 3600:>10.2f}  {errors}"
+		)
+	if not rows:
+		print("запросов не было")
+
+
+def main() -> None:
+	parser = argparse.ArgumentParser(prog="python -m db.cli", description="Клиенты Robotex и их API-ключи")
+	sub = parser.add_subparsers(dest="command", required=True)
+	sub.add_parser("create", help="выпустить ключ новому клиенту").add_argument("name")
+	sub.add_parser("list", help="список клиентов")
+	sub.add_parser("revoke", help="отозвать ключ").add_argument("id", type=int)
+	sub.add_parser("stats", help="статистика использования по клиентам").add_argument("--days", type=float, default=7)
+	asyncio.run(run(parser.parse_args()))
+
+
+if __name__ == "__main__":
+	main()

src/db/clients.py

@@ -0,0 +1,48 @@
+"""Клиенты и API-ключи (#4). Ключ показывается один раз при выпуске, в базе — только sha256."""
+import hashlib
+import secrets
+
+from sqlalchemy import select, update
+
+from db.engine import Database
+from db.models import Client, utcnow
+
+KEY_PREFIX = "rbx_"
+
+
+def hash_key(key: str) -> str:
+	return hashlib.sha256(key.encode()).hexdigest()
+
+
+async def create_client(db: Database, name: str) -> tuple[Client, str]:
+	key = KEY_PREFIX + secrets.token_urlsafe(32)
+	client = Client(name=name, key_hash=hash_key(key), key_prefix=key[:12])
+	async with db.session() as session, session.begin():
+		session.add(client)
+	return client, key
+
+
+async def authenticate(db: Database, key: str | None) -> Client | None:
+	"""Клиент по ключу, если ключ не отозван; отмечает время последнего обращения."""
+	if not key:
+		return None
+	async with db.session() as session, session.begin():
+		client = await session.scalar(
+			select(Client).where(Client.key_hash == hash_key(key), Client.revoked_at.is_(None))
+		)
+		if client:
+			client.last_access_at = utcnow()
+	return client
+
+
+async def revoke(db: Database, client_id: int) -> bool:
+	async with db.session() as session, session.begin():
+		result = await session.execute(
+			update(Client).where(Client.id == client_id, Client.revoked_at.is_(None)).values(revoked_at=utcnow())
+		)
+	return result.rowcount > 0
+
+
+async def list_clients(db: Database) -> list[Client]:
+	async with db.session() as session:
+		return list(await session.scalars(select(Client).order_by(Client.id)))

src/db/engine.py

@@ -0,0 +1,30 @@
+import os
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
+
+from db.models import Base
+
+DEFAULT_URL = "postgresql+asyncpg://robotex:robotex@localhost:5432/robotex"
+# api и worker создают схему при старте одновременно — в Postgres через advisory lock
+_SCHEMA_LOCK = 780078
+
+
+class Database:
+	def __init__(self, engine: AsyncEngine):
+		self.engine = engine
+		self.session = async_sessionmaker(engine, expire_on_commit=False)
+
+	async def init(self) -> None:
+		"""Создать недостающие таблицы."""
+		async with self.engine.begin() as conn:
+			if conn.dialect.name == "postgresql":
+				await conn.execute(text(f"SELECT pg_advisory_xact_lock({_SCHEMA_LOCK})"))
+			await conn.run_sync(Base.metadata.create_all)
+
+	async def close(self) -> None:
+		await self.engine.dispose()
+
+
+def connect(url: str | None = None) -> Database:
+	return Database(create_async_engine(url or os.getenv("DATABASE_URL", DEFAULT_URL), pool_pre_ping=True))

src/db/journal.py

@@ -0,0 +1,69 @@
+"""
+Журнал воркера (#17): окна браузера и задачи. Ошибка записи не роняет задачу — только предупреждение в лог.
+"""
+import logging
+from datetime import datetime, timezone
+
+from sqlalchemy import update
+
+from db.engine import Database
+from db.models import Request, Window, utcnow
+
+log = logging.getLogger(__name__)
+
+
+class Journal:
+	def __init__(self, db: Database | None):
+		self.db = db
+
+	async def _write(self, what: str, fn):
+		if self.db is None:
+			return None
+		try:
+			async with self.db.session() as session, session.begin():
+				return await fn(session)
+		except Exception:
+			log.warning("journal: %s failed", what, exc_info=True)
+			return None
+
+	async def window_opened(self, client_id: int | None, session_id: str, task_id: str, worker_id: str) -> int | None:
+		async def write(session):
+			window = Window(client_id=client_id, session_id=session_id, task_id=task_id, worker_id=worker_id)
+			session.add(window)
+			await session.flush()
+			return window.id
+		return await self._write("window opened", write)
+
+	async def window_used(self, window_id: int | None) -> None:
+		if window_id is not None:
+			await self._write("window used", lambda s: s.execute(
+				update(Window).where(Window.id == window_id).values(last_access_at=utcnow())
+			))
+
+	async def window_closed(self, window_id: int | None) -> None:
+		if window_id is not None:
+			await self._write("window closed", lambda s: s.execute(
+				update(Window).where(Window.id == window_id).values(closed_at=utcnow())
+			))
+
+	async def request_started(
+		self, task_id: str, client_id: int | None, session_id: str, api: str,
+		created_at: float | None, task_data: dict,
+	) -> None:
+		created = datetime.fromtimestamp(created_at, timezone.utc) if created_at else None
+
+		async def write(session):
+			session.add(Request(
+				task_id=task_id, client_id=client_id, session_id=session_id, api=api,
+				created_at=created, task_data=task_data,
+			))
+		await self._write("request started", write)
+
+	async def request_finished(self, task_id: str, window_id: int | None, result: dict | None) -> None:
+		"""result — ответ задачи; None — задачу не удалось выполнить (status failed)."""
+		values = {"finished_at": utcnow(), "window_id": window_id, "status": "failed"}
+		if result is not None:
+			values.update(status=result.get("status"), code=result.get("code"), duration_ms=result.get("duration_ms"))
+		await self._write("request finished", lambda s: s.execute(
+			update(Request).where(Request.task_id == task_id).values(**values)
+		))

src/db/models.py

@@ -0,0 +1,69 @@
+"""
+Таблицы:
+
+* `clients` — клиенты API («Users» в #17): ключ (только sha256), время создания и последнего обращения.
+* `windows` — окна браузера: когда открыто и закрыто, последнее обращение, клиент, задача, открывшая окно.
+  Из них считается время, которое браузер был активен.
+* `requests` — задачи: создание (постановка в очередь), начало и конец выполнения, итог, данные задачи без
+  секретов (прокси, ключ 2captcha, вводимые значения и значения cookies не пишутся).
+"""
+from datetime import datetime, timezone
+
+from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+def utcnow() -> datetime:
+	return datetime.now(timezone.utc)
+
+
+class Base(DeclarativeBase):
+	type_annotation_map = {datetime: DateTime(timezone=True)}
+
+
+class Client(Base):
+	__tablename__ = "clients"
+
+	id: Mapped[int] = mapped_column(Integer, primary_key=True)
+	name: Mapped[str] = mapped_column(String(200))
+	# sha256 ключа: сам ключ показывается один раз при выпуске и нигде не хранится
+	key_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
+	# начало ключа — чтобы отличать ключи в списке
+	key_prefix: Mapped[str] = mapped_column(String(16))
+	created_at: Mapped[datetime] = mapped_column(default=utcnow)
+	last_access_at: Mapped[datetime | None]
+	revoked_at: Mapped[datetime | None]
+
+
+class Window(Base):
+	__tablename__ = "windows"
+
+	id: Mapped[int] = mapped_column(Integer, primary_key=True)
+	client_id: Mapped[int | None] = mapped_column(ForeignKey("clients.id"), index=True)
+	session_id: Mapped[str] = mapped_column(String(36), index=True)
+	# задача, открывшая окно
+	task_id: Mapped[str] = mapped_column(String(36))
+	worker_id: Mapped[str] = mapped_column(String(100))
+	started_at: Mapped[datetime] = mapped_column(default=utcnow)
+	last_access_at: Mapped[datetime] = mapped_column(default=utcnow)
+	closed_at: Mapped[datetime | None]
+
+
+class Request(Base):
+	__tablename__ = "requests"
+
+	id: Mapped[int] = mapped_column(Integer, primary_key=True)
+	task_id: Mapped[str] = mapped_column(String(36), unique=True)
+	client_id: Mapped[int | None] = mapped_column(ForeignKey("clients.id"), index=True)
+	session_id: Mapped[str] = mapped_column(String(36), index=True)
+	window_id: Mapped[int | None] = mapped_column(ForeignKey("windows.id"))
+	# api: /v1/solve или /v2/tasks
+	api: Mapped[str] = mapped_column(String(8))
+	created_at: Mapped[datetime | None]
+	started_at: Mapped[datetime] = mapped_column(default=utcnow, index=True)
+	finished_at: Mapped[datetime | None]
+	duration_ms: Mapped[int | None]
+	# ok / err / failed; code — как в ответе (challenge_failed, action_failed, ...)
+	status: Mapped[str | None] = mapped_column(String(16))
+	code: Mapped[str | None] = mapped_column(String(32))
+	task_data: Mapped[dict] = mapped_column(JSON)

src/db/stats.py

@@ -0,0 +1,77 @@
+"""Статистика использования API (#17): число запросов, их время и время, которое был открыт браузер."""
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+
+from sqlalchemy import select
+
+from db.engine import Database
+from db.models import Client, Request, Window, utcnow
+
+
+@dataclass
+class ClientUsage:
+	client_id: int | None
+	name: str
+	requests: int = 0
+	ok: int = 0
+	# по code: challenge_failed, action_failed, request_failed; failed — задачу не выполнили
+	errors: dict[str, int] = field(default_factory=dict)
+	durations_ms: list[int] = field(default_factory=list)
+	windows: int = 0
+	browser_seconds: float = 0.0
+
+	@property
+	def avg_ms(self) -> int | None:
+		return int(sum(self.durations_ms) / len(self.durations_ms)) if self.durations_ms else None
+
+	@property
+	def p95_ms(self) -> int | None:
+		if not self.durations_ms:
+			return None
+		ordered = sorted(self.durations_ms)
+		return ordered[min(len(ordered) - 1, int(len(ordered) * 0.95))]
+
+
+def _aware(dt: datetime) -> datetime:
+	# sqlite отдаёт время без зоны
+	return dt if dt.tzinfo else dt.replace(tzinfo=utcnow().tzinfo)
+
+
+async def usage(db: Database, since: datetime) -> list[ClientUsage]:
+	"""По клиентам за период: запросы, начатые после since, и время окон в пределах периода."""
+	now = utcnow()
+	async with db.session() as session:
+		names = {c.id: c.name for c in await session.scalars(select(Client))}
+		requests = list(await session.scalars(select(Request).where(Request.started_at >= since)))
+		windows = list(await session.scalars(
+			select(Window).where((Window.closed_at.is_(None)) | (Window.closed_at >= since))
+		))
+
+	result: dict[int | None, ClientUsage] = {}
+
+	def entry(client_id):
+		return result.setdefault(client_id, ClientUsage(client_id, names.get(client_id, "—")))
+
+	for r in requests:
+		u = entry(r.client_id)
+		u.requests += 1
+		if r.status == "ok":
+			u.ok += 1
+		elif r.finished_at is not None:
+			key = r.code or r.status or "unknown"
+			u.errors[key] = u.errors.get(key, 0) + 1
+		if r.duration_ms is not None:
+			u.durations_ms.append(r.duration_ms)
+
+	for w in windows:
+		u = entry(w.client_id)
+		u.windows += 1
+		start = max(_aware(w.started_at), since)
+		end = _aware(w.closed_at) if w.closed_at else now
+		u.browser_seconds += max(0.0, (end - start).total_seconds())
+
+	return sorted(result.values(), key=lambda u: -u.requests)
+
+
+def since_days(days: float) -> datetime:
+	return utcnow() - timedelta(days=days)

src/main.py

@@ -4,16 +4,21 @@
 
 * POST /v2/tasks, GET /v2/tasks/{id}, DELETE /v2/sessions/current — асинхронный API (#80, #82).
 * POST /v1/solve — прежний синхронный контракт поверх той же очереди (#84).
+
+Все эндпоинты, кроме /health, — только с ключом клиента в X-API-Key (#4); ключи — в Postgres (db.cli).
+Задача и окно сессии видны только клиенту, который их создал.
 """
 import asyncio
 import logging
 import os
 import time
 import uuid
+from contextlib import asynccontextmanager
 from typing import Annotated
 
-from fastapi import Cookie, FastAPI, HTTPException, Response
+from fastapi import Cookie, Depends, FastAPI, HTTPException, Response, Security
 from fastapi.responses import JSONResponse
+from fastapi.security import APIKeyHeader
 from starlette.concurrency import run_in_threadpool
 
 logging.basicConfig(
@@ -22,6 +27,9 @@ logging.basicConfig(
 )
 
 from api.schemas import SolveRequest, SolveResponse, TaskCreated, TaskRequest, TaskResponse  # noqa: E402
+from db import clients  # noqa: E402
+from db.engine import connect as connect_db  # noqa: E402
+from db.models import Client  # noqa: E402
 from tasks import broker as queue  # noqa: E402
 from tasks.sessions import new_session_id, parse_session_id  # noqa: E402
 from tasks.store import Status, TaskStore, connect  # noqa: E402
@@ -34,8 +42,28 @@ V1_QUEUE_WAIT = float(os.getenv("V1_QUEUE_WAIT", "30"))
 POLL_INTERVAL = 0.2
 
 store = TaskStore(connect())
+database = connect_db()
 
-app = FastAPI()
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+	await database.init()
+	yield
+	await database.close()
+
+
+app = FastAPI(lifespan=lifespan)
+api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
+
+
+async def current_client(key: Annotated[str | None, Security(api_key_header)]) -> Client:
+	client = await clients.authenticate(database, key)
+	if client is None:
+		raise HTTPException(401, "invalid or missing X-API-Key")
+	return client
+
+
+ClientDep = Annotated[Client, Depends(current_client)]
 
 NO_SLOTS = {"status": "err", "code": "no_slots", "error": "No slots available"}
 
@@ -59,14 +87,15 @@ async def dispatch(message) -> None:
 	await run_in_threadpool(queue.broker.enqueue, message)
 
 
-async def submit(payload: TaskRequest, session_id: str, window: dict | None) -> str:
+async def submit(payload: TaskRequest, session_id: str, window: dict | None, client: Client, api: str) -> str:
 	"""Зарегистрировать задачу и поставить её в очередь: в персональную очередь воркера, где открыто
 	окно сессии (#83), иначе — в общую."""
 	task_id = str(uuid.uuid4())
-	await store.create(task_id, session_id, payload.url)
+	await store.create(task_id, session_id, payload.url, client.id)
 	target = queue.personal_queue(window["worker_id"]) if window else queue.COMMON_QUEUE
 	await dispatch(queue.task_message(
 		task_id, session_id, payload.model_dump(mode="json"), payload.browser_idle_timeout, target,
+		meta={"client_id": client.id, "api": api},
 	))
 	logger.info("task %s [%s] -> %s: %s", task_id, session_id, target, payload.url or "<current page>")
 	return task_id
@@ -76,6 +105,12 @@ async def free_slots() -> int:
 	return sum(w["free_slots"] for w in await store.workers())
 
 
+async def client_window(session_id: str, client: Client) -> dict | None:
+	"""Открытое окно сессии, если оно этого клиента. Чужое окно — как отсутствующее."""
+	window = await store.live_window(session_id)
+	return window if window and window.get("client_id") == str(client.id) else None
+
+
 @app.get("/")
 @app.get("/health")
 async def health():
@@ -92,32 +127,32 @@ async def health():
 async def create_task(
 	payload: TaskRequest,
 	response: Response,
+	client: ClientDep,
 	session_id: Annotated[str | None, Cookie()] = None,
 ):
-	# TODO: добавить поддержку api ключей (#4)
 	session_id = use_session(response, session_id)
-	window = await store.live_window(session_id)
+	window = await client_window(session_id, client)
 	if payload.url is None and window is None:
 		raise HTTPException(422, "url is required: the session has no open browser window")
-	task_id = await submit(payload, session_id, window)
+	task_id = await submit(payload, session_id, window, client, "v2")
 	return {"task_id": task_id, "status": Status.QUEUED, "url": f"/v2/tasks/{task_id}"}
 
 
 @app.get("/v2/tasks/{task_id}", response_model=TaskResponse, response_model_exclude_none=True)
-async def get_task(task_id: str, session_id: Annotated[str | None, Cookie()] = None):
-	# задача видна только своей сессии; чужая и несуществующая неразличимы
+async def get_task(task_id: str, client: ClientDep, session_id: Annotated[str | None, Cookie()] = None):
+	# задача видна только своему клиенту и своей сессии; чужая и несуществующая неразличимы
 	task = await store.get(task_id) if parse_session_id(task_id) else None
-	if not task or task["session_id"] != parse_session_id(session_id):
+	if not task or task["session_id"] != parse_session_id(session_id) or task.get("client_id") != str(client.id):
 		raise HTTPException(404, "task not found")
 	return {"task_id": task_id, "status": task["status"], "error": task.get("error"), "result": task.get("result")}
 
 
 @app.delete("/v2/sessions/current")
-async def close_session(session_id: Annotated[str | None, Cookie()] = None):
+async def close_session(client: ClientDep, session_id: Annotated[str | None, Cookie()] = None):
 	"""Закрыть окно сессии досрочно и освободить слот."""
 	session_id = parse_session_id(session_id)
 	window = await store.get_session(session_id) if session_id else None
-	if not window:
+	if not window or window.get("client_id") != str(client.id):
 		return {"closed": False}
 	await store.delete_session(session_id)
 	if await store.worker_alive(window["worker_id"]):
@@ -129,19 +164,19 @@ async def close_session(session_id: Annotated[str | None, Cookie()] = None):
 async def solve(
 	payload: SolveRequest,
 	response: Response,
+	client: ClientDep,
 	session_id: Annotated[str | None, Cookie()] = None,
 ):
 	"""Синхронный контракт v1: задача ставится в ту же очередь, ответ — когда воркер её выполнит.
 	Если свободных слотов нет или воркер не взял задачу за V1_QUEUE_WAIT — 503, как и раньше."""
-	# TODO: добавить поддержку api ключей (#4)
 	started = time.monotonic()
 	session_id = use_session(response, session_id)
-	window = await store.live_window(session_id)
+	window = await client_window(session_id, client)
 	if window is None and await free_slots() <= 0:
 		logger.warning("no free browser slots, rejecting %s", payload.url)
 		return no_slots_response()
 
-	task_id = await submit(TaskRequest.model_validate(payload.model_dump()), session_id, window)
+	task_id = await submit(TaskRequest.model_validate(payload.model_dump()), session_id, window, client, "v1")
 	while True:
 		task = await store.get(task_id)
 		if task is None:

src/tasks/broker.py

@@ -39,12 +39,13 @@ dramatiq.set_broker(broker)
 
 def task_message(
 	task_id: str, session_id: str, payload: dict, idle_timeout: int, queue: str = COMMON_QUEUE,
+	meta: dict | None = None,
 ) -> dramatiq.Message:
 	"""payload — SolveRequest/TaskRequest.model_dump(): в нём прокси и ключ 2captcha, поэтому
-	он живёт только в сообщении, а не в Redis."""
+	он живёт только в сообщении, а не в Redis. meta — для журнала: client_id, api (v1/v2)."""
 	return dramatiq.Message(
 		queue_name=queue, actor_name=RUN_TASK,
-		args=(task_id, session_id, payload, idle_timeout), kwargs={}, options={},
+		args=(task_id, session_id, payload, idle_timeout, meta or {}), kwargs={}, options={},
 	)
 
 

src/tasks/store.py

@@ -1,9 +1,9 @@
 """
 Состояние задач, окон и воркеров в Redis (#78). Сроки жизни — TTL ключей Redis.
 
-* `robotex:task:<id>` (hash) — статус задачи, сессия, время, результат. Пока задача не завершена,
+* `robotex:task:<id>` (hash) — статус задачи, сессия, клиент, время, результат. Пока задача не завершена,
   у ключа страховочный TTL `PENDING_TTL`; при завершении — `RESULT_TTL` (`RESULT_TTL_HOURS`).
-* `robotex:session:<id>` (hash) — где открыто окно сессии: `worker_id` и `idle_timeout`.
+* `robotex:session:<id>` (hash) — где открыто окно сессии: `worker_id`, `idle_timeout`, `client_id`.
   Пока задача выполняется, TTL — `RUNNING_TTL`; после неё — `browser_idle_timeout`. Ключ исчез —
   окно закрывается (worker/windows.py).
 * `robotex:worker:<id>` — heartbeat воркера с TTL `HEARTBEAT_TTL`: слоты и число окон.
@@ -62,12 +62,13 @@ class TaskStore:
 
 	# --- задачи ---
 
-	async def create(self, task_id: str, session_id: str, url: str | None) -> None:
+	async def create(self, task_id: str, session_id: str, url: str | None, client_id: int | None = None) -> None:
 		key = _task_key(task_id)
 		async with self.redis.pipeline(transaction=True) as pipe:
 			pipe.hset(key, mapping={
 				"status": Status.QUEUED,
 				"session_id": session_id,
+				"client_id": "" if client_id is None else client_id,
 				"url": url or "",
 				"created_at": time.time(),
 				"attempts": 0,
@@ -156,10 +157,14 @@ class TaskStore:
 
 	# --- окна сессий ---
 
-	async def set_session(self, session_id: str, worker_id: str, idle_timeout: int, ttl: int) -> None:
+	async def set_session(
+		self, session_id: str, worker_id: str, idle_timeout: int, ttl: int, client_id: int | None = None,
+	) -> None:
 		key = _session_key(session_id)
 		async with self.redis.pipeline(transaction=True) as pipe:
-			pipe.hset(key, mapping={"worker_id": worker_id, "idle_timeout": idle_timeout})
+			pipe.hset(key, mapping={
+				"worker_id": worker_id, "idle_timeout": idle_timeout, "client_id": "" if client_id is None else client_id,
+			})
 			pipe.expire(key, ttl)
 			await pipe.execute()
 

src/worker/actors.py

@@ -19,6 +19,8 @@ import dramatiq  # noqa: E402
 from dramatiq.asyncio import get_event_loop_thread  # noqa: E402
 from dramatiq.middleware import Middleware  # noqa: E402
 
+from db.engine import connect as connect_db  # noqa: E402
+from db.journal import Journal  # noqa: E402
 from tasks import store as task_store  # noqa: E402
 from tasks.broker import CLOSE_WINDOW, COMMON_QUEUE, RUN_TASK, broker, personal_queue  # noqa: E402
 from worker.windows import Outcome, WindowPool  # noqa: E402
@@ -65,7 +67,9 @@ async def _cleanup_loop(pool: WindowPool) -> None:
 
 async def _start() -> None:
 	global _pool
-	_pool = WindowPool(task_store.TaskStore(task_store.connect()), WORKER_ID, BROWSER_SLOTS)
+	db = connect_db()
+	await db.init()
+	_pool = WindowPool(task_store.TaskStore(task_store.connect()), WORKER_ID, BROWSER_SLOTS, Journal(db))
 	await _pool.report()
 	_background.extend([asyncio.create_task(_heartbeat_loop(_pool)), asyncio.create_task(_cleanup_loop(_pool))])
 	log.info("worker %s: %d browser slots, queues %s, %s", WORKER_ID, BROWSER_SLOTS, COMMON_QUEUE, personal_queue(WORKER_ID))
@@ -77,6 +81,8 @@ async def _stop() -> None:
 	_background.clear()
 	await _pool.close_all()
 	await _pool.store.remove_worker(WORKER_ID)
+	if _pool.journal.db:
+		await _pool.journal.db.close()
 
 
 class Lifecycle(Middleware):
@@ -95,10 +101,10 @@ broker.declare_queue(personal_queue(WORKER_ID))
 
 
 @dramatiq.actor(actor_name=RUN_TASK, queue_name=COMMON_QUEUE, max_retries=0, time_limit=TASK_TIME_LIMIT_MS)
-def run_task(task_id: str, session_id: str, payload: dict, idle_timeout: int) -> None:
-	if _run(_pool.handle(task_id, session_id, payload, idle_timeout)) == Outcome.NO_SLOT:
+def run_task(task_id: str, session_id: str, payload: dict, idle_timeout: int, meta: dict | None = None) -> None:
+	if _run(_pool.handle(task_id, session_id, payload, idle_timeout, meta)) == Outcome.NO_SLOT:
 		# в общую очередь: окно могло открыться на любом воркере, где есть слот
-		run_task.send_with_options(args=(task_id, session_id, payload, idle_timeout), delay=NO_SLOT_DELAY_MS)
+		run_task.send_with_options(args=(task_id, session_id, payload, idle_timeout, meta), delay=NO_SLOT_DELAY_MS)
 
 
 @dramatiq.actor(actor_name=CLOSE_WINDOW, queue_name=COMMON_QUEUE, max_retries=0)

src/worker/windows.py

@@ -29,8 +29,9 @@ from patchright.async_api import Page, ProxySettings
 
 from api import actions, profile
 from api.geoip import resolve_geo
-from api.logs import safe_request_dump, safe_response_dump
+from api.logs import safe_request_data, safe_request_dump, safe_response_dump
 from api.schemas import TaskRequest
+from db.journal import Journal
 from engine import AsyncStealthySession
 from tasks.sessions import parse_session_id
 from tasks.store import RUNNING_TTL, Status, TaskStore
@@ -64,6 +65,9 @@ class Window:
 	proxy: str | None = None
 	# окно закрыть, как только закончится текущая задача (DELETE /v2/sessions/current)
 	closing: bool = False
+	client_id: int | None = None
+	# строка окна в журнале (#17): открыто — закрыто
+	journal_id: int | None = None
 
 
 @dataclass
@@ -158,8 +162,9 @@ def build_result(payload: TaskRequest, result: dict) -> dict:
 
 
 class WindowPool:
-	def __init__(self, store: TaskStore, worker_id: str, slots: int):
+	def __init__(self, store: TaskStore, worker_id: str, slots: int, journal: Journal | None = None):
 		self.store = store
+		self.journal = journal or Journal(None)
 		self.worker_id = worker_id
 		self.slots = slots
 		self.used = 0
@@ -198,8 +203,13 @@ class WindowPool:
 		entry = self._locks.get(session_id)
 		return bool(entry and entry.lock.locked())
 
-	async def handle(self, task_id: str, session_id: str, payload: dict, idle_timeout: int) -> str:
-		"""Выполнить задачу. Результат — в Redis; возврат — что делать с сообщением очереди."""
+	async def handle(
+		self, task_id: str, session_id: str, payload: dict, idle_timeout: int, meta: dict | None = None,
+	) -> str:
+		"""Выполнить задачу. Результат — в Redis; возврат — что делать с сообщением очереди.
+		meta: client_id и api (v1/v2) — для журнала."""
+		meta = meta or {}
+		client_id = meta.get("client_id")
 		async with self._session_lock(session_id):
 			window = self.windows.get(session_id)
 			new_window = window is None
@@ -210,7 +220,7 @@ class WindowPool:
 					return Outcome.NO_SLOT if task and task["status"] == Status.QUEUED else Outcome.SKIPPED
 				# слот занимается до первого await, иначе его может перехватить другая задача
 				self.used += 1
-				window = self.windows[session_id] = Window(session_id)
+				window = self.windows[session_id] = Window(session_id, client_id=client_id)
 				await self.report()
 
 			if not await self.store.claim(task_id, self.worker_id, running_here=task_id in self.running):
@@ -219,9 +229,16 @@ class WindowPool:
 				return Outcome.SKIPPED
 
 			self.running.add(task_id)
+			task = TaskRequest.model_validate(payload)
+			result = None
+			queued = await self.store.get(task_id) or {}
+			await self.journal.request_started(
+				task_id, client_id, session_id, meta.get("api", "v2"),
+				float(queued["created_at"]) if queued.get("created_at") else None, safe_request_data(task),
+			)
 			try:
-				await self.store.set_session(session_id, self.worker_id, idle_timeout, RUNNING_TTL)
-				result = await self._execute(window, TaskRequest.model_validate(payload))
+				await self.store.set_session(session_id, self.worker_id, idle_timeout, RUNNING_TTL, client_id)
+				result = await self._execute(window, task, task_id)
 				await self.store.finish(task_id, result)
 			except Exception as e:
 				log.exception("task %s failed", task_id)
@@ -234,15 +251,16 @@ class WindowPool:
 				raise
 			finally:
 				self.running.discard(task_id)
+				await self.journal.request_finished(task_id, window.journal_id, result)
 				if idle_timeout and window.engine and not window.closing:
-					await self.store.set_session(session_id, self.worker_id, idle_timeout, idle_timeout * 60)
+					await self.store.set_session(session_id, self.worker_id, idle_timeout, idle_timeout * 60, client_id)
 				else:
 					await self._close(window)
 					with suppress(Exception):
 						await self.store.delete_session(session_id)
 		return Outcome.DONE
 
-	async def _execute(self, window: Window, payload: TaskRequest) -> dict:
+	async def _execute(self, window: Window, payload: TaskRequest, task_id: str) -> dict:
 		started = time.monotonic()
 		session_id = window.session_id
 		log.info("Incoming request [%s]: %s", session_id, payload.url or "<current page>")
@@ -264,11 +282,14 @@ class WindowPool:
 					raise NoWindow("url is required: the session has no open browser window")
 				window.engine = await self._open(session_id, payload, nav_kwargs)
 				window.proxy = payload.proxy
-			elif payload.cookies is not None:
-				# переданные cookies заменяют cookies профиля, как и при запуске окна
-				await window.engine.context.clear_cookies()
-				if payload.cookies:
-					await window.engine.context.add_cookies(payload.cookies)
+				window.journal_id = await self.journal.window_opened(window.client_id, session_id, task_id, self.worker_id)
+			else:
+				await self.journal.window_used(window.journal_id)
+				if payload.cookies is not None:
+					# переданные cookies заменяют cookies профиля, как и при запуске окна
+					await window.engine.context.clear_cookies()
+					if payload.cookies:
+						await window.engine.context.add_cookies(payload.cookies)
 
 			if payload.url:
 				raw = await window.engine.fetch(
@@ -324,6 +345,8 @@ class WindowPool:
 
 	async def _close_engine(self, window: Window) -> None:
 		engine, window.engine = window.engine, None
+		journal_id, window.journal_id = window.journal_id, None
+		await self.journal.window_closed(journal_id)
 		if engine is not None:
 			try:
 				await engine.close()