feat(seed): derive created_at/updated_at from git history

app/seed.py

@@ -17,7 +17,8 @@
 
 import hashlib
 import json
-from datetime import date
+import subprocess
+from datetime import date, datetime
 from pathlib import Path
 from typing import Any
 
@@ -39,6 +40,43 @@
 DATA_DIR = get_data_root()
 
 
+NUL = "\x00"  # git log record separator, so a commit line cannot look like a path
+
+
+def _git_timestamps(data_dir: Path) -> dict[str, tuple[datetime, datetime]]:
+    """Map each record path to (first commit, last commit) times.
+
+    `created_at`/`updated_at` used to be stamped with "now" at seed time, so
+    every dump rewrote every page with a timestamp that only said when the dump
+    ran. Git already knows when a record appeared and when it last changed, and
+    those answers do not move between runs.
+
+    One `git log` pass over the whole data tree; an empty map (no git, shallow
+    clone) leaves the model defaults in place.
+    """
+    try:
+        log = subprocess.run(
+            # --relative keeps paths relative to data_dir, matching __path.
+            ["git", "log", "--reverse", "--no-renames", "--relative",
+             "--format=%x00%cI", "--name-only", "--diff-filter=AM", "--", "."],
+            cwd=data_dir, check=True, capture_output=True, text=True, encoding="utf-8",
+        ).stdout
+    except (OSError, subprocess.CalledProcessError):
+        return {}
+
+    stamps: dict[str, tuple[datetime, datetime]] = {}
+    when: datetime | None = None
+    for line in log.splitlines():
+        if line.startswith(NUL):
+            when = datetime.fromisoformat(line[1:])
+            continue
+        if when is None or not line.endswith(".json"):
+            continue
+        first, _ = stamps.get(line, (when, when))
+        stamps[line] = (first, when)
+    return stamps
+
+
 def _load_dir(subdir: Path) -> list[dict[str, Any]]:
     if not subdir.exists():
         return []
@@ -52,6 +90,7 @@ def _load_dir(subdir: Path) -> list[dict[str, Any]]:
         for key, value in list(record.items()):
             if key.endswith("_date") and isinstance(value, str):
                 record[key] = date.fromisoformat(value)
+        record["__path"] = path.relative_to(subdir.parent).as_posix()
         items.append(record)
     return items
 
@@ -82,6 +121,27 @@ def _with_id(obj: Any, taken: set[int]) -> Any:
     return obj
 
 
+def _stamp(obj: Any, path: str | None, stamps: dict[str, tuple[datetime, datetime]]) -> Any:
+    """Replace the seed-time timestamps with the record's git history."""
+    times = stamps.get(path or "")
+    # brand and gpu carry no timestamps at all — skip rather than invent fields.
+    if times and hasattr(obj, "created_at"):
+        obj.created_at, obj.updated_at = times
+    return obj
+
+
+def _row(
+    model: Any,
+    record: dict[str, Any],
+    taken: set[int],
+    stamps: dict[str, tuple[datetime, datetime]],
+    **fks: Any,
+) -> Any:
+    """Build one row: resolved FKs, stable id, git-derived timestamps."""
+    path = record.pop("__path", None)
+    return _stamp(_with_id(model(**fks, **record), taken), path, stamps)
+
+
 def _existing_slugs(session: Session, model: type[SQLModel]) -> set[str]:
     rows = session.exec(select(model)).all()
     return {row.slug for row in rows}  # type: ignore[attr-defined]  # all data models have slug
@@ -91,6 +151,7 @@ def seed(session: Session, data_dir: Path = DATA_DIR) -> dict[str, int]:
     """Idempotently insert seed data. Returns counts of newly inserted rows."""
     # Ids assigned in this run; only used to break hash collisions.
     taken: set[int] = set()
+    stamps = _git_timestamps(data_dir)
     counts = {
         "brands": 0,
         "socs": 0,
@@ -114,7 +175,7 @@ def seed(session: Session, data_dir: Path = DATA_DIR) -> dict[str, int]:
         # `categories` lives in the JSON for browsing/validation only — the Brand
         # table model does not (yet) carry it, so drop before construction.
         record.pop("categories", None)
-        session.add(_with_id(Brand(**record), taken))
+        session.add(_row(Brand, record, taken, stamps))
         counts["brands"] += 1
     session.commit()
 
@@ -131,7 +192,7 @@ def seed(session: Session, data_dir: Path = DATA_DIR) -> dict[str, int]:
             raise ValueError(
                 f"SoC '{record['slug']}' references unknown brand '{manufacturer}'"
             )
-        session.add(_with_id(SoC(manufacturer_id=manufacturer_id, **record), taken))
+        session.add(_row(SoC, record, taken, stamps, manufacturer_id=manufacturer_id))
         counts["socs"] += 1
     session.commit()
 
@@ -154,8 +215,7 @@ def seed(session: Session, data_dir: Path = DATA_DIR) -> dict[str, int]:
             raise ValueError(
                 f"Smartphone '{record['slug']}' references unknown SoC '{soc_slug}'"
             )
-        phone = Smartphone(brand_id=brand_id, soc_id=soc_id, **record)
-        session.add(_with_id(phone, taken))
+        session.add(_row(Smartphone, record, taken, stamps, brand_id=brand_id, soc_id=soc_id))
         counts["smartphones"] += 1
     session.commit()
 
@@ -178,8 +238,7 @@ def seed_mobile_devices(subdir: str, model: type[SQLModel], count_key: str) -> N
                     f"{subdir.rstrip('s').title()} '{record['slug']}' "
                     f"references unknown SoC '{soc_slug}'"
                 )
-            device = model(brand_id=brand_id, soc_id=soc_id, **record)
-            session.add(_with_id(device, taken))
+            session.add(_row(model, record, taken, stamps, brand_id=brand_id, soc_id=soc_id))
             counts[count_key] += 1
         session.commit()
 
@@ -198,7 +257,7 @@ def seed_mobile_devices(subdir: str, model: type[SQLModel], count_key: str) -> N
             raise ValueError(
                 f"GPU '{record['slug']}' references unknown brand '{manufacturer}'"
             )
-        session.add(_with_id(DiscreteGPU(manufacturer_id=manufacturer_id, **record), taken))
+        session.add(_row(DiscreteGPU, record, taken, stamps, manufacturer_id=manufacturer_id))
         counts["gpus"] += 1
     session.commit()
 
@@ -213,7 +272,7 @@ def seed_mobile_devices(subdir: str, model: type[SQLModel], count_key: str) -> N
             raise ValueError(
                 f"CPU '{record['slug']}' references unknown brand '{manufacturer}'"
             )
-        session.add(_with_id(CPU(manufacturer_id=manufacturer_id, **record), taken))
+        session.add(_row(CPU, record, taken, stamps, manufacturer_id=manufacturer_id))
         counts["cpus"] += 1
     session.commit()
 
@@ -242,8 +301,9 @@ def seed_mobile_devices(subdir: str, model: type[SQLModel], count_key: str) -> N
             raise ValueError(
                 f"Laptop '{record['slug']}' references unknown GPU '{gpu_slug}'"
             )
-        laptop = Laptop(brand_id=brand_id, cpu_id=cpu_id, gpu_id=gpu_id, **record)
-        session.add(_with_id(laptop, taken))
+        session.add(
+            _row(Laptop, record, taken, stamps, brand_id=brand_id, cpu_id=cpu_id, gpu_id=gpu_id)
+        )
         counts["laptops"] += 1
     session.commit()
 
@@ -258,7 +318,7 @@ def seed_mobile_devices(subdir: str, model: type[SQLModel], count_key: str) -> N
             raise ValueError(
                 f"Monitor '{record['slug']}' references unknown brand '{brand_slug}'"
             )
-        session.add(_with_id(Monitor(brand_id=brand_id, **record), taken))
+        session.add(_row(Monitor, record, taken, stamps, brand_id=brand_id))
         counts["monitors"] += 1
     session.commit()
 
@@ -267,7 +327,7 @@ def seed_mobile_devices(subdir: str, model: type[SQLModel], count_key: str) -> N
     for record in _load_dir(data_dir / "software"):
         if record["slug"] in software_slugs:
             continue
-        session.add(_with_id(Software(**record), taken))
+        session.add(_row(Software, record, taken, stamps))
         counts["software"] += 1
     session.commit()
 
@@ -276,7 +336,7 @@ def seed_mobile_devices(subdir: str, model: type[SQLModel], count_key: str) -> N
     for record in _load_dir(data_dir / "website"):
         if record["slug"] in website_slugs:
             continue
-        session.add(_with_id(Website(**record), taken))
+        session.add(_row(Website, record, taken, stamps))
         counts["websites"] += 1
     session.commit()
 

tests/unit/test_git_timestamps.py

@@ -0,0 +1,66 @@
+"""Timestamps come from git, so a re-dump of unchanged data is byte-identical."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+from pathlib import Path
+
+from app.seed import _git_timestamps
+
+
+def _git(repo: Path, *args: str, when: str | None = None) -> None:
+    env = {
+        "GIT_AUTHOR_NAME": "t",
+        "GIT_AUTHOR_EMAIL": "t@example.com",
+        "GIT_COMMITTER_NAME": "t",
+        "GIT_COMMITTER_EMAIL": "t@example.com",
+        "PATH": os.environ["PATH"],
+    }
+    if when:
+        # _git_timestamps reads committer dates, so both must be pinned.
+        env["GIT_AUTHOR_DATE"] = env["GIT_COMMITTER_DATE"] = when
+    subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True, env=env)
+
+
+def _repo(tmp_path: Path) -> Path:
+    data = tmp_path / "data"
+    (data / "cpu").mkdir(parents=True)
+    _git(tmp_path, "init", "-q")
+    return data
+
+
+def _commit(repo: Path, rel: str, payload: dict, when: str) -> None:
+    path = repo / rel
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    _git(repo.parent, "add", "-A")
+    _git(repo.parent, "commit", "-q", "-m", rel, when=when)
+
+
+def test_created_and_updated_track_first_and_last_commit(tmp_path):
+    data = _repo(tmp_path)
+    _commit(data, "cpu/a.json", {"slug": "a"}, "2026-01-02T03:04:05+00:00")
+    _commit(data, "cpu/b.json", {"slug": "b"}, "2026-02-02T03:04:05+00:00")
+    _commit(data, "cpu/a.json", {"slug": "a", "cores": 8}, "2026-03-02T03:04:05+00:00")
+
+    stamps = _git_timestamps(data)
+    created_a, updated_a = stamps["cpu/a.json"]
+    created_b, updated_b = stamps["cpu/b.json"]
+
+    assert created_a < updated_a          # edited later
+    assert created_b == updated_b         # written once
+    assert created_a < created_b          # a came first
+
+
+def test_repeated_reads_agree(tmp_path):
+    data = _repo(tmp_path)
+    _commit(data, "cpu/a.json", {"slug": "a"}, "2026-01-02T03:04:05+00:00")
+    assert _git_timestamps(data) == _git_timestamps(data)
+
+
+def test_outside_a_git_repo_returns_nothing(tmp_path):
+    plain = tmp_path / "plain"
+    (plain / "cpu").mkdir(parents=True)
+    assert _git_timestamps(plain) == {}