file_last_commits.py
python
sha256:92528ae07d0e1239d87fd5fd1f439e8fbb49c9778a9a400bc4a736073fb28316
feat: byte-range blob reads, file attribution DAG walk, bra…
Sonnet 4.6
minor
⚠ breaking
43 days ago
| 1 | """Materialized per-file last-commit computation. |
| 2 | |
| 3 | Called at push time to pre-compute which commit last touched each file. |
| 4 | The result is stored in musehub_file_last_commits so page-load reads are |
| 5 | a single indexed SQL query instead of an O(N × blob) walk. |
| 6 | """ |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | from datetime import timezone as _tz |
| 10 | |
| 11 | from sqlalchemy import select |
| 12 | from sqlalchemy.dialects.postgresql import insert as pg_insert |
| 13 | from sqlalchemy.ext.asyncio import AsyncSession |
| 14 | |
| 15 | from musehub.db.musehub_intel_models import MusehubFileLastCommit |
| 16 | from musehub.db.musehub_repo_models import MusehubCommit |
| 17 | from musehub.types.json_types import StrDict |
| 18 | |
| 19 | _DAG_BATCH = 200 |
| 20 | |
| 21 | |
| 22 | async def _walk_dag(session: AsyncSession, head_commit_id: str) -> list[MusehubCommit]: |
| 23 | """BFS walk of the commit DAG from head_commit_id, newest→oldest. |
| 24 | |
| 25 | Uses parent_ids to follow ancestry rather than filtering by branch name, |
| 26 | so historical commits pushed on a different branch are correctly included. |
| 27 | """ |
| 28 | visited: set[str] = set() |
| 29 | frontier: list[str] = [head_commit_id] |
| 30 | ordered: list[MusehubCommit] = [] |
| 31 | |
| 32 | while frontier: |
| 33 | batch = [cid for cid in frontier if cid not in visited] |
| 34 | if not batch: |
| 35 | break |
| 36 | frontier = [] |
| 37 | for i in range(0, len(batch), _DAG_BATCH): |
| 38 | chunk = batch[i : i + _DAG_BATCH] |
| 39 | result = await session.execute( |
| 40 | select(MusehubCommit).where(MusehubCommit.commit_id.in_(chunk)) |
| 41 | ) |
| 42 | for commit in result.scalars().all(): |
| 43 | if commit.commit_id not in visited: |
| 44 | visited.add(commit.commit_id) |
| 45 | ordered.append(commit) |
| 46 | for parent_id in (commit.parent_ids or []): |
| 47 | if parent_id and parent_id not in visited: |
| 48 | frontier.append(parent_id) |
| 49 | |
| 50 | # Sort newest→oldest by timestamp for the snapshot diff walk. |
| 51 | ordered.sort(key=lambda c: c.timestamp, reverse=True) |
| 52 | return ordered |
| 53 | |
| 54 | |
| 55 | async def compute_and_store_file_last_commits( |
| 56 | session: AsyncSession, |
| 57 | repo_id: str, |
| 58 | branch: str, |
| 59 | head_commit_id: str, |
| 60 | ) -> None: |
| 61 | """Walk commits from head_commit_id via parent_ids and upsert file attribution rows. |
| 62 | |
| 63 | Uses DAG ancestry rather than branch-name filtering so historical commits |
| 64 | pushed on a different branch (e.g. main) are correctly included when |
| 65 | computing attribution for dev. |
| 66 | """ |
| 67 | commits = await _walk_dag(session, head_commit_id) |
| 68 | if not commits: |
| 69 | return |
| 70 | |
| 71 | from musehub.services.musehub_snapshot import get_snapshot_manifests_batch, _MAX_BATCH_SIZE |
| 72 | snap_ids = [c.snapshot_id for c in commits if c.snapshot_id] |
| 73 | if not snap_ids: |
| 74 | return |
| 75 | |
| 76 | snap_by_id: dict[str, StrDict] = {} |
| 77 | for i in range(0, len(snap_ids), _MAX_BATCH_SIZE): |
| 78 | snap_by_id.update(await get_snapshot_manifests_batch(session, snap_ids[i : i + _MAX_BATCH_SIZE])) |
| 79 | |
| 80 | # Walk newest→oldest to find per-file attribution. |
| 81 | file_attribution: dict[str, MusehubCommit] = {} |
| 82 | prev_manifest: dict[str, str] = {} |
| 83 | prev_commit: MusehubCommit | None = None |
| 84 | |
| 85 | for commit in commits: |
| 86 | cur_manifest = snap_by_id.get(commit.snapshot_id or "", {}) |
| 87 | if prev_commit is not None: |
| 88 | for path, prev_oid in prev_manifest.items(): |
| 89 | if path not in file_attribution: |
| 90 | cur_oid = cur_manifest.get(path) |
| 91 | if cur_oid != prev_oid: |
| 92 | file_attribution[path] = prev_commit |
| 93 | prev_manifest = cur_manifest |
| 94 | prev_commit = commit |
| 95 | |
| 96 | # Files still in the oldest manifest are attributed to the oldest commit. |
| 97 | if prev_commit is not None: |
| 98 | for path in prev_manifest: |
| 99 | if path not in file_attribution: |
| 100 | file_attribution[path] = prev_commit |
| 101 | |
| 102 | if not file_attribution: |
| 103 | return |
| 104 | |
| 105 | # Upsert: INSERT ... ON CONFLICT DO UPDATE. |
| 106 | rows = [] |
| 107 | for path, commit in file_attribution.items(): |
| 108 | ts = commit.timestamp |
| 109 | if ts.tzinfo is None: |
| 110 | ts = ts.replace(tzinfo=_tz.utc) |
| 111 | rows.append({ |
| 112 | "repo_id": repo_id, |
| 113 | "branch": branch, |
| 114 | "path": path, |
| 115 | "commit_id": commit.commit_id, |
| 116 | "commit_message": (commit.message or "").split("\n")[0][:72], |
| 117 | "commit_author": commit.author or "", |
| 118 | "commit_timestamp": ts, |
| 119 | "agent_id": commit.agent_id or None, |
| 120 | "model_id": commit.model_id or None, |
| 121 | }) |
| 122 | |
| 123 | stmt = pg_insert(MusehubFileLastCommit).values(rows) |
| 124 | stmt = stmt.on_conflict_do_update( |
| 125 | index_elements=["repo_id", "branch", "path"], |
| 126 | set_={ |
| 127 | "commit_id": stmt.excluded.commit_id, |
| 128 | "commit_message": stmt.excluded.commit_message, |
| 129 | "commit_author": stmt.excluded.commit_author, |
| 130 | "commit_timestamp": stmt.excluded.commit_timestamp, |
| 131 | "agent_id": stmt.excluded.agent_id, |
| 132 | "model_id": stmt.excluded.model_id, |
| 133 | }, |
| 134 | ) |
| 135 | await session.execute(stmt) |
File History
1 commit
sha256:92528ae07d0e1239d87fd5fd1f439e8fbb49c9778a9a400bc4a736073fb28316
feat: byte-range blob reads, file attribution DAG walk, bra…
Sonnet 4.6
minor
⚠
43 days ago