feat: manage external source lifecycle

apps/api/alembic/versions/0016_source_lifecycle.py

@@ -0,0 +1,24 @@
+"""track source record lifecycle checks
+
+Revision ID: 0016
+Revises: 0015
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0016"
+down_revision = "0015"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("external_observation", sa.Column("source_check_status", sa.String(length=30)))
+    op.add_column("external_observation", sa.Column("source_checked_at", sa.DateTime(timezone=True)))
+
+
+def downgrade() -> None:
+    op.drop_column("external_observation", "source_checked_at")
+    op.drop_column("external_observation", "source_check_status")

apps/api/app/community_importer.py

@@ -68,6 +68,7 @@ def stage_observations(
             "weight_g": _integer(payload.get("weight_g"), minimum=1, maximum=3_000_000),
             "published_at": _datetime(payload.get("published_at")),
             "last_seen_at": fetched_at, "payload": payload,
+            "source_check_status": "available", "source_checked_at": fetched_at,
         }
         if observation is None:
             observation = ExternalObservation(
@@ -89,8 +90,17 @@ def stage_observations(
                 observation.fish = None
                 observation.waterbody = None
                 observation.review_note = "Source record changed; manual mapping and publication required"
+                observation.reviewed_at = fetched_at
+                observation.moderation_version += 1
             for key, value in values.items():
                 setattr(observation, key, value)
+            if observation.status == "withdrawn":
+                observation.status = "staged"
+                observation.fish = None
+                observation.waterbody = None
+                observation.review_note = "Source record reappeared; manual confirmation required"
+                observation.reviewed_at = fetched_at
+                observation.moderation_version += 1
             updated += 1
         touched.append(observation)
     session.commit()

apps/api/app/models.py

@@ -199,6 +199,8 @@ class ExternalObservation(Base):
     review_note: Mapped[str | None] = mapped_column(Text)
     reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     moderation_version: Mapped[int] = mapped_column(default=0)
+    source_check_status: Mapped[str | None] = mapped_column(String(30))
+    source_checked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     source: Mapped[DataSource] = relationship()
     fish: Mapped[Fish | None] = relationship()
     waterbody: Mapped[Waterbody | None] = relationship()

apps/api/app/routers/admin.py

@@ -156,13 +156,14 @@ def _external_out(item: ExternalObservation) -> ExternalObservationOut:
         catch_report_id=item.catch_report_id, review_note=item.review_note,
         missing_fields=missing_fields, source_payload=allowed_payload,
         moderation_version=item.moderation_version,
+        source_check_status=item.source_check_status, source_checked_at=item.source_checked_at,
     )
 
 
 @router.get("/api/v1/admin/external-observations", response_model=list[ExternalObservationOut])
 def admin_external_observations(
     db: Db, _: Annotated[str, Depends(_admin)],
-    status: Literal["staged", "mapped", "ready", "published", "rejected", "review"] | None = None,
+    status: Literal["staged", "mapped", "ready", "published", "rejected", "withdrawn", "review"] | None = None,
     source_system: str | None = None,
     completeness: Literal["all", "complete", "incomplete"] = "all",
     order: Literal["newest", "oldest", "risk"] = "newest",
@@ -328,4 +329,3 @@ def delete_report(report_id: UUID, db: Db, moderator: Annotated[str, Depends(_ad
     db.commit()
     public_cache.invalidate()
     return Response(status_code=204)
-

apps/api/app/routers/public_data.py

@@ -40,7 +40,7 @@ def community_observations(
         joinedload(ExternalObservation.source)
     ).where(
         ExternalObservation.catch_report_id.is_(None),
-        ExternalObservation.status != "rejected",
+        ExternalObservation.status.not_in(["rejected", "withdrawn"]),
         DataSource.enabled.is_(True),
     )
     if waterbody:

apps/api/app/schemas.py

@@ -239,6 +239,8 @@ class ExternalObservationOut(BaseModel):
     missing_fields: list[str]
     source_payload: dict[str, str | int | float | bool | None]
     moderation_version: int
+    source_check_status: str | None
+    source_checked_at: datetime | None
 
 
 class ExternalObservationMapping(BaseModel):

apps/api/app/source_lifecycle.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Literal
+
+from sqlalchemy.orm import Session
+
+from .models import ExternalObservation, ModerationStatus
+
+
+SourceCheckStatus = Literal["available", "missing", "temporary_error", "blocked"]
+
+
+def record_source_check(
+    session: Session,
+    observation: ExternalObservation,
+    status: SourceCheckStatus,
+    *,
+    checked_at: datetime | None = None,
+) -> ExternalObservation:
+    """Persist a check performed during an already scheduled source request.
+
+    Only an authoritative 404/410-style ``missing`` result withdraws published
+    data. Transient errors and access blocks remain diagnostic and never remove
+    an observation from activity.
+    """
+    current = checked_at or datetime.now(timezone.utc)
+    observation.source_check_status = status
+    observation.source_checked_at = current
+    if status == "missing" and observation.status != "withdrawn":
+        if observation.catch_report is not None:
+            observation.catch_report.moderation_status = ModerationStatus.pending
+        observation.status = "withdrawn"
+        observation.review_note = "Source record missing; withdrawn pending moderator review"
+        observation.reviewed_at = current
+        observation.moderation_version += 1
+    session.commit()
+    return observation

apps/api/openapi.json

@@ -820,6 +820,29 @@
             ],
             "title": "Reviewed At"
           },
+          "source_check_status": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Source Check Status"
+          },
+          "source_checked_at": {
+            "anyOf": [
+              {
+                "format": "date-time",
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Source Checked At"
+          },
           "source_external_id": {
             "title": "Source External Id",
             "type": "string"
@@ -942,7 +965,9 @@
           "review_note",
           "missing_fields",
           "source_payload",
-          "moderation_version"
+          "moderation_version",
+          "source_check_status",
+          "source_checked_at"
         ],
         "title": "ExternalObservationOut",
         "type": "object"
@@ -2109,6 +2134,7 @@
                     "ready",
                     "published",
                     "rejected",
+                    "withdrawn",
                     "review"
                   ],
                   "type": "string"

apps/api/tests/test_community_importer.py

@@ -9,6 +9,7 @@ from sqlalchemy.orm import Session
 
 from app.community_importer import CommunityImportError, stage_observations
 from app.community_review import ExternalReviewError, map_observation, publish_observation, suggest_aliases
+from app.source_lifecycle import record_source_check
 from app.database import Base
 from app.models import CatchReport, DataSource, ExternalEntityAlias, ExternalObservation, Fish, Waterbody
 from rf4_research.community_sources import parse_rf4db_catches, parse_rf4map_point, parse_rf4posts_spot
@@ -128,6 +129,8 @@ def test_changed_published_record_requires_review_and_reuses_report(db: Session)
     assert report.moderation_status.value == "pending"
     assert report.weight_g == 5000
     assert item.weight_g == 6000
+    assert item.moderation_version == 1
+    assert item.reviewed_at is not None
     stage_observations(db, [record() | {"weight_g": 6000}])
     assert item.status == "staged"
     with pytest.raises(ExternalReviewError):
@@ -140,6 +143,49 @@ def test_changed_published_record_requires_review_and_reuses_report(db: Session)
     assert db.scalar(select(func.count()).select_from(CatchReport)) == 1
 
 
+def test_missing_source_withdraws_published_record_until_manual_review(db: Session) -> None:
+    fish = Fish(slug="pike", name_ru="Щука")
+    water = Waterbody(slug="test-lake", name_ru="Тестовое озеро")
+    db.add_all([fish, water])
+    db.commit()
+    seen = datetime(2026, 9, 13, 8, tzinfo=timezone.utc)
+    checked = datetime(2026, 9, 13, 9, tzinfo=timezone.utc)
+    stage_observations(db, [record() | {"weight_g": 5000}], fetched_at=seen)
+    item = db.scalar(select(ExternalObservation))
+    assert item is not None and item.catch_report is not None
+
+    record_source_check(db, item, "missing", checked_at=checked)
+
+    assert item.status == "withdrawn"
+    assert item.source_check_status == "missing"
+    assert item.source_checked_at.replace(tzinfo=timezone.utc) == checked
+    assert item.catch_report.moderation_status.value == "pending"
+    assert item.moderation_version == 1
+
+    stage_observations(db, [record() | {"weight_g": 5000}], fetched_at=checked)
+    assert item.status == "staged"
+    assert item.source_check_status == "available"
+    assert item.catch_report.moderation_status.value == "pending"
+    assert "reappeared" in (item.review_note or "")
+
+
+@pytest.mark.parametrize("status", ["temporary_error", "blocked"])
+def test_non_authoritative_source_failures_do_not_withdraw(db: Session, status: str) -> None:
+    fish = Fish(slug="pike", name_ru="Щука")
+    water = Waterbody(slug="test-lake", name_ru="Тестовое озеро")
+    db.add_all([fish, water])
+    db.commit()
+    stage_observations(db, [record() | {"weight_g": 5000}])
+    item = db.scalar(select(ExternalObservation))
+    assert item is not None and item.catch_report is not None
+
+    record_source_check(db, item, status)  # type: ignore[arg-type]
+
+    assert item.status == "published"
+    assert item.catch_report.moderation_status.value == "approved"
+    assert item.source_check_status == status
+
+
 def test_auto_publication_requires_enabled_source(db: Session) -> None:
     source = DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=False)
     fish = Fish(slug="pike", name_ru="Щука")