fix(admin-jobs): UPSERT в PUT /admin/jobs/settings — не терять изменения (#1223)

UPDATE без проверки rowcount затрагивал 0 строк для несуществующего
job_type; get_one возвращал hardcoded _fallback. PUT отвечал 200,
админ считал настройку сохранённой — но она терялась (новый job_type
до seed-миграции, опечатка в path).

Patch: update() проверяет result.rowcount: при 0 делает INSERT с
переданными колонками (непереданные → дефолты таблицы из м.81).
INSERT использует ON CONFLICT (job_type) DO NOTHING + повторный
UPDATE как защита от гонки. Поведение для существующих строк не
изменилось. 8 новых тестов (6 service + 2 API).

Closes #1223

backend/app/services/job_settings.py

@@ -192,10 +192,30 @@ def update(
     extra_config: dict | None = None,
     updated_by: str | None = None,
 ) -> dict[str, Any]:
-    """PATCH-style update — обновляет только переданные поля.
+    """PATCH-style upsert — обновляет только переданные поля.
+
+    Если строки с таким job_type ещё нет, она создаётся (с дефолтами таблицы из
+    data/sql/81_job_settings.sql для непереданных колонок). Раньше UPDATE без
+    проверки rowcount молча терял изменения для несуществующих job_type, а PUT
+    отвечал 200 c fallback-дефолтами (issue #1223).
 
     cron_schedule='' → NULL (manual-only). updated_at выставляется в NOW().
     """
+    import json
+
+    # Нормализация значений по той же логике, что и в исходном UPDATE.
+    # cron_schedule: None = не трогать; '' = явный NULL (manual-only)
+    cron_provided = cron_schedule is not None
+    norm_cron: str | None = cron_schedule if cron_schedule else None
+
+    # rate_ms: None = не трогать; 0 = явно обнулить (no throttle)
+    rate_provided = rate_ms is not None
+    norm_rate: int | None = rate_ms if rate_provided and rate_ms and rate_ms > 0 else None
+
+    extra_provided = extra_config is not None
+    norm_extra_json = json.dumps(extra_config, ensure_ascii=False) if extra_provided else None
+
+    # ── UPDATE для существующей строки (PATCH-семантика без изменений) ───────
     sets: list[str] = ["updated_at = NOW()"]
     params: dict[str, Any] = {"jt": job_type}
 
@@ -210,26 +230,81 @@ def update(
             sets.append(f"{field} = :{field}")
             params[field] = value
 
-    # cron_schedule: None = не трогать; '' = явный NULL (manual-only)
-    if cron_schedule is not None:
+    if cron_provided:
         sets.append("cron_schedule = :cron_schedule")
-        params["cron_schedule"] = cron_schedule if cron_schedule else None
+        params["cron_schedule"] = norm_cron
 
-    # rate_ms: None = не трогать; 0 = явно обнулить (no throttle)
-    if rate_ms is not None:
+    if rate_provided:
         sets.append("rate_ms = :rate_ms")
-        params["rate_ms"] = rate_ms if rate_ms > 0 else None
+        params["rate_ms"] = norm_rate
 
-    if extra_config is not None:
+    if extra_provided:
         sets.append("extra_config = CAST(:extra_config AS jsonb)")
-        import json
+        params["extra_config"] = norm_extra_json
 
-        params["extra_config"] = json.dumps(extra_config, ensure_ascii=False)
-
-    db.execute(
+    result = db.execute(
         text(f"UPDATE job_settings SET {', '.join(sets)} WHERE job_type = :jt"),
         params,
     )
+
+    # Если строки не было — INSERT с явными значениями только для переданных колонок.
+    # Остальные колонки получат defaults таблицы (enabled=true, queue_name='celery',
+    # max_retries=2, max_concurrency=1, extra_config='{}'::jsonb, updated_at=now()).
+    if (result.rowcount or 0) == 0:
+        insert_cols: list[str] = ["job_type"]
+        insert_vals: list[str] = [":jt"]
+        insert_params: dict[str, Any] = {"jt": job_type}
+
+        for field, value in [
+            ("enabled", enabled),
+            ("queue_name", queue_name),
+            ("max_retries", max_retries),
+            ("max_concurrency", max_concurrency),
+            ("updated_by", updated_by),
+        ]:
+            if value is not None:
+                insert_cols.append(field)
+                insert_vals.append(f":{field}")
+                insert_params[field] = value
+
+        if cron_provided:
+            insert_cols.append("cron_schedule")
+            insert_vals.append(":cron_schedule")
+            insert_params["cron_schedule"] = norm_cron
+
+        if rate_provided:
+            insert_cols.append("rate_ms")
+            insert_vals.append(":rate_ms")
+            insert_params["rate_ms"] = norm_rate
+
+        if extra_provided:
+            insert_cols.append("extra_config")
+            insert_vals.append("CAST(:extra_config AS jsonb)")
+            insert_params["extra_config"] = norm_extra_json
+
+        # ON CONFLICT DO NOTHING — защита от гонки: между нашими UPDATE и INSERT
+        # параллельный PUT мог вставить строку. В этом случае повторим UPDATE,
+        # чтобы наши изменения не потерялись.
+        insert_result = db.execute(
+            text(
+                f"INSERT INTO job_settings ({', '.join(insert_cols)}) "
+                f"VALUES ({', '.join(insert_vals)}) "
+                "ON CONFLICT (job_type) DO NOTHING"
+            ),
+            insert_params,
+        )
+        if (insert_result.rowcount or 0) == 0:
+            logger.info(
+                "job_settings.update: гонка по '%s' — повторный UPDATE после INSERT race",
+                job_type,
+            )
+            db.execute(
+                text(f"UPDATE job_settings SET {', '.join(sets)} WHERE job_type = :jt"),
+                params,
+            )
+        else:
+            logger.info("job_settings.update: вставлена новая строка '%s'", job_type)
+
     db.commit()
     return get_one(job_type, db)
 

backend/tests/api/v1/test_admin_jobs_settings.py

@@ -0,0 +1,136 @@
+"""API-тесты для PUT /api/v1/admin/jobs/settings/{job_type}.
+
+Регрессия по issue #1223: PUT для отсутствующей строки молча терял изменения
+и отвечал 200 с fallback-дефолтами. После фикса update() выполняет INSERT,
+если UPDATE затронул 0 строк, и возвращает новую строку с сохранёнными полями.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+
+def _make_db_recorder(update_rowcounts: list[int]) -> tuple[MagicMock, list[tuple[str, Any]]]:
+    """Db-mock, который:
+    - возвращает заданные rowcount для UPDATE/INSERT (по списку),
+    - имитирует поведение get_one(): SELECT возвращает строку с теми же
+      значениями, что были последним INSERT (или None, если не вставлялась).
+    """
+    db = MagicMock()
+    calls: list[tuple[str, Any]] = []
+    state: dict[str, Any] = {"inserted": None}
+    plan = list(update_rowcounts)
+
+    def execute(stmt: Any, params: Any = None) -> Any:
+        sql = str(stmt)
+        calls.append((sql, params))
+        result = MagicMock()
+        if "UPDATE job_settings" in sql:
+            result.rowcount = plan.pop(0) if plan else 0
+        elif "INSERT INTO job_settings" in sql:
+            result.rowcount = plan.pop(0) if plan else 0
+            if result.rowcount > 0 and isinstance(params, dict):
+                state["inserted"] = params
+        elif "SELECT" in sql and "job_settings" in sql:
+            # Имитация get_one после успешного INSERT
+            row = state["inserted"]
+            if row is None:
+                result.mappings.return_value.first.return_value = None
+            else:
+                import datetime
+
+                result.mappings.return_value.first.return_value = {
+                    "job_type": row.get("jt"),
+                    "enabled": row.get("enabled", True),
+                    "queue_name": row.get("queue_name", "celery"),
+                    "cron_schedule": row.get("cron_schedule"),
+                    "rate_ms": row.get("rate_ms"),
+                    "max_retries": row.get("max_retries", 2),
+                    "max_concurrency": row.get("max_concurrency", 1),
+                    "extra_config": {},  # JSON-строка сериализована — для теста плоский dict ok
+                    "updated_at": datetime.datetime(2026, 6, 13, tzinfo=datetime.UTC),
+                    "updated_by": row.get("updated_by"),
+                    "description": None,
+                }
+        return result
+
+    db.execute = execute
+    db.commit = MagicMock()
+    return db, calls
+
+
+def _override_db(db: MagicMock) -> Any:
+    def _factory():
+        yield db
+
+    return _factory
+
+
+def test_put_job_settings_existing_row_updates_only() -> None:
+    """Существующая строка — UPDATE, без INSERT."""
+    from app.core.db import get_db
+
+    db, calls = _make_db_recorder(update_rowcounts=[1])
+    app.dependency_overrides[get_db] = _override_db(db)
+
+    try:
+        client = TestClient(app)
+        r = client.put(
+            "/api/v1/admin/jobs/settings/scrape_kn",
+            json={"rate_ms": 750},
+        )
+        assert r.status_code == 200, r.text
+        assert any("UPDATE job_settings" in sql for sql, _ in calls)
+        assert not any("INSERT INTO job_settings" in sql for sql, _ in calls)
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_put_job_settings_missing_row_inserts_and_returns_persisted_values() -> None:
+    """Issue #1223: PUT для отсутствующей строки должен сохранить изменения.
+
+    После фикса: UPDATE → 0 строк, затем INSERT с переданными полями.
+    Ответ должен содержать **переданные** значения, а не fallback-дефолты.
+    """
+    from app.core.db import get_db
+
+    # UPDATE 0, INSERT 1 — типовой happy-path для несуществующей строки.
+    db, calls = _make_db_recorder(update_rowcounts=[0, 1])
+    app.dependency_overrides[get_db] = _override_db(db)
+
+    try:
+        client = TestClient(app)
+        r = client.put(
+            "/api/v1/admin/jobs/settings/brand_new_job",
+            json={
+                "enabled": False,
+                "rate_ms": 1234,
+                "max_retries": 5,
+            },
+        )
+        assert r.status_code == 200, r.text
+        body = r.json()
+        # Ключевая проверка: ответ — НЕ fallback (_fallback ставит enabled=True).
+        # Значит INSERT действительно произошёл, get_one вернул вставленную строку.
+        assert body["job_type"] == "brand_new_job"
+        assert body["enabled"] is False
+        assert body["rate_ms"] == 1234
+        assert body["max_retries"] == 5
+
+        # Подтверждаем, что INSERT реально был.
+        assert any("INSERT INTO job_settings" in sql for sql, _ in calls)
+        insert_call = next(
+            (sql, params) for sql, params in calls if "INSERT INTO job_settings" in sql
+        )
+        assert "ON CONFLICT (job_type) DO NOTHING" in insert_call[0]
+        assert insert_call[1]["jt"] == "brand_new_job"
+        assert insert_call[1]["enabled"] is False
+        assert insert_call[1]["rate_ms"] == 1234
+        assert insert_call[1]["updated_by"] == "admin-ui"
+    finally:
+        app.dependency_overrides.clear()

backend/tests/services/test_job_settings.py

@@ -0,0 +1,186 @@
+"""Unit-тесты для app.services.job_settings.update (UPSERT поведение).
+
+Покрывают issue #1223: PUT /admin/jobs/settings/{job_type} для несуществующей
+строки молча терял изменения. После фикса update() выполняет INSERT, если
+UPDATE затронул 0 строк.
+
+DB здесь полностью замокана — мы проверяем, что service:
+  1. Делает UPDATE сначала и фиксирует rowcount.
+  2. При rowcount == 0 делает INSERT с переданными колонками.
+  3. При rowcount > 0 НЕ делает INSERT (PATCH-семантика сохранена).
+  4. Защищён от гонки UPDATE→INSERT через ON CONFLICT DO NOTHING.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.services import job_settings as svc
+
+
+class _ExecRecorder:
+    """Имитация db.execute() с задаваемым rowcount и записью всех вызовов."""
+
+    def __init__(self, plan: list[int]) -> None:
+        # plan: список rowcount-ов для последовательных execute() вызовов.
+        # get_one() в конце update() тоже бьёт execute() — этот вызов покрывается
+        # последним элементом или дефолтным MagicMock (см. __call__).
+        self._plan = list(plan)
+        self.calls: list[tuple[str, dict[str, Any]]] = []
+
+    def __call__(self, stmt: Any, params: Any = None) -> Any:
+        sql = str(stmt)
+        self.calls.append((sql, dict(params) if isinstance(params, dict) else {}))
+        result = MagicMock()
+        # rowcount только для DML; для SELECT/get_one — не важен.
+        result.rowcount = self._plan.pop(0) if self._plan else 0
+        # get_one() ожидает .mappings().first() → None (→ fallback).
+        result.mappings.return_value.first.return_value = None
+        return result
+
+
+@pytest.fixture
+def _patched_get_one(monkeypatch: pytest.MonkeyPatch) -> MagicMock:
+    """get_one() имеет свою БД-логику; патчим, чтобы не мешала assert-ам на execute()."""
+    mock = MagicMock(return_value={"job_type": "x", "stub": True})
+    monkeypatch.setattr(svc, "get_one", mock)
+    return mock
+
+
+def _count_statements(calls: list[tuple[str, dict[str, Any]]], keyword: str) -> int:
+    return sum(1 for sql, _ in calls if keyword in sql)
+
+
+def test_update_existing_row_only_updates(_patched_get_one: MagicMock) -> None:
+    """Когда UPDATE затронул 1 строку — INSERT НЕ выполняется (regression guard)."""
+    db = MagicMock()
+    exec_rec = _ExecRecorder(plan=[1])  # UPDATE → 1 row affected
+    db.execute = exec_rec
+
+    svc.update("scrape_kn", db, rate_ms=500, updated_by="admin-ui")
+
+    assert _count_statements(exec_rec.calls, "UPDATE job_settings") == 1
+    assert _count_statements(exec_rec.calls, "INSERT INTO job_settings") == 0
+    db.commit.assert_called_once()
+
+
+def test_update_missing_row_inserts_with_provided_fields(
+    _patched_get_one: MagicMock,
+) -> None:
+    """Issue #1223: PUT для несуществующей строки должен сохранять изменения.
+
+    UPDATE возвращает rowcount=0, после чего service делает INSERT с теми же
+    колонками, что были в SET-списке.
+    """
+    db = MagicMock()
+    exec_rec = _ExecRecorder(plan=[0, 1])  # UPDATE 0, INSERT 1
+    db.execute = exec_rec
+
+    svc.update(
+        "new_job_type",
+        db,
+        enabled=False,
+        queue_name="celery",
+        cron_schedule="*/5 * * * *",
+        rate_ms=1000,
+        max_retries=3,
+        max_concurrency=2,
+        extra_config={"key": "value"},
+        updated_by="admin-ui",
+    )
+
+    # Должен быть ровно один UPDATE и ровно один INSERT.
+    assert _count_statements(exec_rec.calls, "UPDATE job_settings") == 1
+    assert _count_statements(exec_rec.calls, "INSERT INTO job_settings") == 1
+
+    insert_sql, insert_params = next(
+        (sql, params) for sql, params in exec_rec.calls if "INSERT" in sql
+    )
+    assert "ON CONFLICT (job_type) DO NOTHING" in insert_sql
+    assert insert_params["jt"] == "new_job_type"
+    assert insert_params["enabled"] is False
+    assert insert_params["queue_name"] == "celery"
+    assert insert_params["cron_schedule"] == "*/5 * * * *"
+    assert insert_params["rate_ms"] == 1000
+    assert insert_params["max_retries"] == 3
+    assert insert_params["max_concurrency"] == 2
+    assert insert_params["updated_by"] == "admin-ui"
+    # extra_config сериализуется в JSON-строку — приводится через CAST.
+    assert "CAST(:extra_config AS jsonb)" in insert_sql
+    assert '"key": "value"' in insert_params["extra_config"]
+
+    db.commit.assert_called_once()
+
+
+def test_update_missing_row_minimal_payload(_patched_get_one: MagicMock) -> None:
+    """INSERT для пустого payload (только updated_by из роута) — лишь job_type/updated_by.
+
+    Остальные колонки должны получить таблицы-дефолты — мы их не упоминаем в
+    INSERT, проверяем что они НЕ попали в params.
+    """
+    db = MagicMock()
+    exec_rec = _ExecRecorder(plan=[0, 1])
+    db.execute = exec_rec
+
+    svc.update("brand_new", db, updated_by="admin-ui")
+
+    insert_sql, insert_params = next(
+        (sql, params) for sql, params in exec_rec.calls if "INSERT" in sql
+    )
+    assert insert_params == {"jt": "brand_new", "updated_by": "admin-ui"}
+    # В INSERT не должно быть колонок, которые мы не передали.
+    assert "enabled" not in insert_sql
+    assert "queue_name" not in insert_sql
+    assert "extra_config" not in insert_sql
+
+
+def test_update_insert_race_falls_back_to_second_update(
+    _patched_get_one: MagicMock,
+) -> None:
+    """Гонка UPDATE→INSERT: параллельный PUT вставил строку между нашими запросами.
+
+    INSERT ... ON CONFLICT DO NOTHING вернёт rowcount=0; service должен повторить
+    UPDATE, чтобы наши изменения не потерялись.
+    """
+    db = MagicMock()
+    # UPDATE→0, INSERT→0 (конфликт), second UPDATE→1.
+    exec_rec = _ExecRecorder(plan=[0, 0, 1])
+    db.execute = exec_rec
+
+    svc.update("racing_jt", db, enabled=True, updated_by="admin-ui")
+
+    assert _count_statements(exec_rec.calls, "UPDATE job_settings") == 2
+    assert _count_statements(exec_rec.calls, "INSERT INTO job_settings") == 1
+    db.commit.assert_called_once()
+
+
+def test_update_cron_empty_string_becomes_null(_patched_get_one: MagicMock) -> None:
+    """cron_schedule='' → NULL и при UPDATE, и при INSERT (manual-only режим)."""
+    db = MagicMock()
+    # UPDATE→0 → попадаем в INSERT-ветку.
+    exec_rec = _ExecRecorder(plan=[0, 1])
+    db.execute = exec_rec
+
+    svc.update("nspd_geo", db, cron_schedule="", updated_by="admin-ui")
+
+    update_params = exec_rec.calls[0][1]
+    insert_params = exec_rec.calls[1][1]
+    assert update_params["cron_schedule"] is None
+    assert insert_params["cron_schedule"] is None
+
+
+def test_update_rate_ms_zero_becomes_null(_patched_get_one: MagicMock) -> None:
+    """rate_ms=0 → NULL (no throttle)."""
+    db = MagicMock()
+    exec_rec = _ExecRecorder(plan=[0, 1])
+    db.execute = exec_rec
+
+    svc.update("nspd_geo", db, rate_ms=0, updated_by="admin-ui")
+
+    update_params = exec_rec.calls[0][1]
+    insert_params = exec_rec.calls[1][1]
+    assert update_params["rate_ms"] is None
+    assert insert_params["rate_ms"] is None