diff --git a/application/database/db.py b/application/database/db.py index 4ca0dc040..1969e9e9b 100644 --- a/application/database/db.py +++ b/application/database/db.py @@ -8,6 +8,7 @@ import time import yaml +from datetime import datetime, timezone from pprint import pprint from collections import Counter, defaultdict @@ -257,6 +258,108 @@ class StagedChangeSet(BaseModel): # type: ignore created_at = sqla.Column(sqla.DateTime, nullable=False) +class ArtifactIngestEvent(BaseModel): # type: ignore + """Tracks one harvested artifact persisted per import run.""" + + __tablename__ = "artifact_ingest_event" + id = sqla.Column(sqla.String, primary_key=True, default=generate_uuid) + run_id = sqla.Column( + sqla.String, + sqla.ForeignKey("import_run.id", onupdate="CASCADE", ondelete="CASCADE"), + nullable=False, + ) + artifact_id = sqla.Column(sqla.String, nullable=False) + harvest_mode = sqla.Column(sqla.String, nullable=False) + event_type = sqla.Column(sqla.String, nullable=False) + source_json = sqla.Column(sqla.Text, nullable=False) + locator_json = sqla.Column(sqla.Text, nullable=False) + artifact_json = sqla.Column(sqla.Text, nullable=False) + harvest_json = sqla.Column(sqla.Text, nullable=False) + observed_at = sqla.Column(sqla.DateTime, nullable=False) + created_at = sqla.Column(sqla.DateTime, nullable=False) + + __table_args__ = ( + sqla.UniqueConstraint( + run_id, + artifact_id, + name="uq_artifact_ingest_event_run_artifact", + ), + ) + + +class IngestChunk(BaseModel): # type: ignore + """Tracks every chunk belonging to an artifact ingest event.""" + + __tablename__ = "ingest_chunk" + id = sqla.Column(sqla.String, primary_key=True, default=generate_uuid) + artifact_event_id = sqla.Column( + sqla.String, + sqla.ForeignKey( + "artifact_ingest_event.id", + onupdate="CASCADE", + ondelete="CASCADE", + ), + nullable=False, + ) + chunk_id = sqla.Column(sqla.String, nullable=False) + text = sqla.Column(sqla.Text, nullable=False) + char_count = sqla.Column(sqla.Integer, nullable=False) + span_json = sqla.Column(sqla.Text, nullable=False) + delta_json = sqla.Column(sqla.Text, nullable=True) + created_at = sqla.Column(sqla.DateTime, nullable=False) + + __table_args__ = ( + sqla.UniqueConstraint( + artifact_event_id, + chunk_id, + name="uq_ingest_chunk_artifact_chunk", + ), + ) + + +class HarvesterCheckpoint(BaseModel): # type: ignore + __tablename__ = "harvester_checkpoint" + repository_id = sqla.Column(sqla.String, primary_key=True) + provider = sqla.Column(sqla.String, nullable=False) + owner = sqla.Column(sqla.String, nullable=False) + repository = sqla.Column(sqla.String, nullable=False) + branch = sqla.Column(sqla.String, nullable=False) + last_processed_commit = sqla.Column(sqla.String, nullable=True) + created_at = sqla.Column( + sqla.DateTime(timezone=True), + nullable=False, + default=lambda: datetime.now(timezone.utc), + ) + updated_at = sqla.Column( + sqla.DateTime(timezone=True), + nullable=False, + default=lambda: datetime.now(timezone.utc), + ) + __table_args__ = ( + sqla.UniqueConstraint( + "provider", + "owner", + "repository", + "branch", + name="uq_harvester_checkpoint_canonical_source", + ), + ) + + +def _serialize_json_value(value: Any) -> str: + return flask_json.dumps(value) + + +def _normalize_utc_datetime(value: Any) -> Any: + from datetime import datetime, timezone + + if isinstance(value, datetime): + if value.tzinfo is None: + return value + return value.astimezone(timezone.utc) + return value + + def create_import_run(source: str, version: Optional[str] = None) -> ImportRun: """Create and persist an import run record. Returns the new ImportRun.""" from datetime import datetime, timezone @@ -304,6 +407,68 @@ def get_previous_import_run(source: str, current_run_id: str) -> Optional[Import ) +def create_artifact_ingest_event( + *, + run_id: str, + artifact_id: str, + harvest_mode: str, + event_type: str, + source_json: Any, + locator_json: Any, + artifact_json: Any, + harvest_json: Any, + observed_at: Any, +) -> ArtifactIngestEvent: + from datetime import datetime, timezone + + observed_at = _normalize_utc_datetime(observed_at) + + event = ArtifactIngestEvent( + id=generate_uuid(), + run_id=run_id, + artifact_id=artifact_id, + harvest_mode=harvest_mode, + event_type=event_type, + source_json=_serialize_json_value(source_json), + locator_json=_serialize_json_value(locator_json), + artifact_json=_serialize_json_value(artifact_json), + harvest_json=_serialize_json_value(harvest_json), + observed_at=observed_at, + created_at=_normalize_utc_datetime(datetime.now(timezone.utc)), + ) + sqla.session.add(event) + sqla.session.commit() + return event + + +def create_ingest_chunk( + *, + artifact_event_id: str, + chunk_id: str, + text: str, + char_count: int, + span_json: Any, + delta_json: Optional[Any] = None, +) -> IngestChunk: + from datetime import datetime, timezone + + chunk = IngestChunk( + id=generate_uuid(), + artifact_event_id=artifact_event_id, + chunk_id=chunk_id, + text=text, + char_count=char_count, + span_json=_serialize_json_value(span_json), + delta_json=( + _serialize_json_value(delta_json) if delta_json is not None else None + ), + created_at=_normalize_utc_datetime(datetime.now(timezone.utc)), + ) + sqla.session.add(chunk) + sqla.session.commit() + return chunk + + def persist_standard_snapshot( *, run_id: str, diff --git a/application/tests/harvester_test/change_detector_test.py b/application/tests/harvester_test/change_detector_test.py new file mode 100644 index 000000000..5cb33456a --- /dev/null +++ b/application/tests/harvester_test/change_detector_test.py @@ -0,0 +1,168 @@ +import unittest +from unittest.mock import MagicMock +from unittest.mock import call +from unittest.mock import patch + +from application.utils.harvester.change_detector import ( + ChangeDetector, +) + + +class ChangeDetectorTests(unittest.TestCase): + @patch("application.utils.harvester.change_detector.subprocess.run") + def test_get_modified_files_since(self, mock_run): + client = MagicMock() + client.get_local_path.return_value = "repo-under-test" + + mock_run.side_effect = [ + MagicMock(stdout="resolved_base\n"), + MagicMock(stdout="resolved_target\n"), + MagicMock(stdout="a.md\nb.md\na.md\n"), + ] + + detector = ChangeDetector(client) + + files = detector.get_modified_files_since( + "base", + "target", + ) + + self.assertEqual( + files, + [ + "a.md", + "b.md", + ], + ) + + mock_run.assert_has_calls( + [ + call( + [ + "git", + "-C", + "repo-under-test", + "rev-parse", + "--verify", + "--end-of-options", + "base^{commit}", + ], + capture_output=True, + text=True, + check=True, + timeout=60, + ), + call( + [ + "git", + "-C", + "repo-under-test", + "rev-parse", + "--verify", + "--end-of-options", + "target^{commit}", + ], + capture_output=True, + text=True, + check=True, + timeout=60, + ), + call( + [ + "git", + "-C", + "repo-under-test", + "diff", + "--name-only", + "resolved_base", + "resolved_target", + ], + capture_output=True, + text=True, + check=True, + timeout=60, + ), + ] + ) + + @patch("application.utils.harvester.change_detector.subprocess.run") + def test_get_commits_since(self, mock_run): + client = MagicMock() + client.get_local_path.return_value = "repo-under-test" + + mock_run.side_effect = [ + MagicMock(stdout="resolved_base\n"), + MagicMock(stdout="resolved_target\n"), + MagicMock(stdout="111\n222\n333\n"), + ] + + detector = ChangeDetector(client) + + commits = detector.get_commits_since( + "base", + "target", + ) + + self.assertEqual( + commits, + [ + "111", + "222", + "333", + ], + ) + + self.assertEqual( + mock_run.call_args_list, + [ + call( + [ + "git", + "-C", + "repo-under-test", + "rev-parse", + "--verify", + "--end-of-options", + "base^{commit}", + ], + capture_output=True, + text=True, + check=True, + timeout=60, + ), + call( + [ + "git", + "-C", + "repo-under-test", + "rev-parse", + "--verify", + "--end-of-options", + "target^{commit}", + ], + capture_output=True, + text=True, + check=True, + timeout=60, + ), + call( + [ + "git", + "-C", + "repo-under-test", + "log", + "--reverse", + "--format=%H", + "resolved_base..resolved_target", + ], + capture_output=True, + text=True, + check=True, + timeout=60, + ), + ], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/checkpoint_store_test.py b/application/tests/harvester_test/checkpoint_store_test.py new file mode 100644 index 000000000..5d5f1e947 --- /dev/null +++ b/application/tests/harvester_test/checkpoint_store_test.py @@ -0,0 +1,193 @@ +import unittest +from datetime import datetime, timezone + +from application import create_app, sqla +from application.utils.harvester.checkpoint_store import CheckpointStore +from application.utils.harvester.models import RepositoryCheckpoint + + +class CheckpointStoreTests(unittest.TestCase): + def setUp(self) -> None: + self.app = create_app(mode="test") + self.app_context = self.app.app_context() + self.app_context.push() + sqla.create_all() + + def tearDown(self) -> None: + sqla.session.remove() + sqla.drop_all() + self.app_context.pop() + + def test_save_and_load_checkpoint(self): + store = CheckpointStore() + checkpoint = RepositoryCheckpoint( + repository_id="owasp-asvs", + last_processed_commit="abc123", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="owasp", + repository="asvs", + branch="main", + ) + + store.save(checkpoint) + loaded = store.load("owasp-asvs") + + self.assertIsNotNone(loaded) + assert loaded is not None + self.assertEqual(loaded.last_processed_commit, "abc123") + self.assertEqual(loaded.provider, "github") + + def test_update_upsert_and_two_repositories_remain_isolated(self): + store = CheckpointStore() + repo_a = RepositoryCheckpoint( + repository_id="repo-a", + last_processed_commit="commit-1", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="repo-a", + branch="main", + ) + repo_b = RepositoryCheckpoint( + repository_id="repo-b", + last_processed_commit="commit-b", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="repo-b", + branch="main", + ) + store.save(repo_a) + store.save(repo_b) + + updated_a = RepositoryCheckpoint( + repository_id="repo-a", + last_processed_commit="commit-2", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="repo-a", + branch="main", + ) + store.save(updated_a) + + loaded_a = store.load("repo-a") + loaded_b = store.load("repo-b") + + self.assertIsNotNone(loaded_a) + self.assertIsNotNone(loaded_b) + assert loaded_a is not None + assert loaded_b is not None + self.assertEqual(loaded_a.last_processed_commit, "commit-2") + self.assertEqual(loaded_b.last_processed_commit, "commit-b") + + def test_duplicate_canonical_source_identity_rejected(self): + store = CheckpointStore() + first = RepositoryCheckpoint( + repository_id="repo-a", + last_processed_commit="commit-1", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="shared", + branch="main", + ) + second = RepositoryCheckpoint( + repository_id="repo-b", + last_processed_commit="commit-2", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="shared", + branch="main", + ) + + store.save(first) + + with self.assertRaises(ValueError): + store.save(second) + + def test_immutable_repository_identity(self): + store = CheckpointStore() + first = RepositoryCheckpoint( + repository_id="repo-a", + last_processed_commit="commit-1", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="repo-a", + branch="main", + ) + store.save(first) + + conflicting = RepositoryCheckpoint( + repository_id="repo-a", + last_processed_commit="commit-2", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="repo-a", + branch="develop", + ) + + with self.assertRaises(ValueError): + store.save(conflicting) + + def test_null_initial_checkpoint(self): + store = CheckpointStore() + checkpoint = RepositoryCheckpoint( + repository_id="repo-a", + last_processed_commit=None, + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="repo-a", + branch="main", + ) + + store.save(checkpoint) + loaded = store.load("repo-a") + + self.assertIsNotNone(loaded) + assert loaded is not None + self.assertIsNone(loaded.last_processed_commit) + + def test_transaction_rollback_leaves_previous_checkpoint_intact(self): + store = CheckpointStore() + original = RepositoryCheckpoint( + repository_id="repo-a", + last_processed_commit="commit-1", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="repo-a", + branch="main", + ) + store.save(original) + + conflicting = RepositoryCheckpoint( + repository_id="repo-a", + last_processed_commit="commit-2", + updated_at=datetime.now(timezone.utc), + provider="github", + owner="sample", + repository="repo-a", + branch="develop", + ) + + with self.assertRaises(ValueError): + store.save(conflicting) + + loaded = store.load("repo-a") + self.assertIsNotNone(loaded) + assert loaded is not None + self.assertEqual(loaded.last_processed_commit, "commit-1") + + def test_load_missing_repository(self): + store = CheckpointStore() + self.assertIsNone(store.load("repo-b")) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/git_repository_client_integration_test.py b/application/tests/harvester_test/git_repository_client_integration_test.py index 4803e1d67..f2f585098 100644 --- a/application/tests/harvester_test/git_repository_client_integration_test.py +++ b/application/tests/harvester_test/git_repository_client_integration_test.py @@ -7,6 +7,7 @@ from application.utils.harvester.git_repository_client import ( GitRepositoryClient, ) +from application.utils.harvester.change_detector import ChangeDetector class IntegrationGitRepositoryClient(GitRepositoryClient): @@ -201,6 +202,71 @@ def run_sync(client): self.assertEqual((self.cache / "test.txt").read_text(), "v1") + def test_change_detector_uses_captured_target_sha(self): + client = self.create_client() + client.clone() + + detector = ChangeDetector(client) + + base = client.get_current_commit_sha() + + (self.work / "file.txt").write_text("B") + git("add", ".", cwd=self.work) + git("commit", "-m", "second", cwd=self.work) + git("push", "origin", "main", cwd=self.work) + + client.fetch() + + target = client.get_current_commit_sha() + + (self.work / "another.txt").write_text("C") + git("add", ".", cwd=self.work) + git("commit", "-m", "third", cwd=self.work) + git("push", "origin", "main", cwd=self.work) + + client.fetch() + files = detector.get_modified_files_since(base, target) + commits = detector.get_commits_since(base, target) + + self.assertEqual(files, ["file.txt"]) + self.assertEqual(commits, [target]) + + def test_change_detector_returns_commits_oldest_first(self): + client = self.create_client() + client.clone() + + detector = ChangeDetector(client) + base = client.get_current_commit_sha() + + (self.work / "file.txt").write_text("B") + git("add", ".", cwd=self.work) + git("commit", "-m", "B", cwd=self.work) + commit_b = git_output("rev-parse", "HEAD", cwd=self.work) + + (self.work / "file.txt").write_text("C") + git("add", ".", cwd=self.work) + git("commit", "-m", "C", cwd=self.work) + commit_c = git_output("rev-parse", "HEAD", cwd=self.work) + + (self.work / "file.txt").write_text("D") + git("add", ".", cwd=self.work) + git("commit", "-m", "D", cwd=self.work) + commit_d = git_output("rev-parse", "HEAD", cwd=self.work) + + git("push", "origin", "main", cwd=self.work) + + client.fetch() + + commits = detector.get_commits_since(base, commit_d) + self.assertEqual( + commits, + [ + commit_b, + commit_c, + commit_d, + ], + ) + if __name__ == "__main__": unittest.main() diff --git a/application/tests/import_run_test.py b/application/tests/import_run_test.py index c3e303625..7b03161a6 100644 --- a/application/tests/import_run_test.py +++ b/application/tests/import_run_test.py @@ -1,6 +1,9 @@ """Tests for import run metadata (Step 6).""" +import json import unittest +from datetime import datetime, timezone + from application import create_app, sqla from application.database import db @@ -31,3 +34,52 @@ def test_get_latest_import_run(self) -> None: self.assertIsNotNone(latest) self.assertEqual(latest.id, run2.id) self.assertEqual(latest.version, "2.0") + + def test_create_artifact_ingest_event_and_chunk(self) -> None: + run = db.create_import_run(source="artifact_ingest", version="1.0") + observed_at = datetime.now(timezone.utc) + + event = db.create_artifact_ingest_event( + run_id=run.id, + artifact_id="artifact-1", + harvest_mode="backfill", + event_type="discovered", + source_json={"uri": "https://example.com/source"}, + locator_json={"path": "/tmp/source"}, + artifact_json={"id": "artifact-1"}, + harvest_json={"status": "ok"}, + observed_at=observed_at, + ) + + self.assertIsNotNone(event.id) + self.assertEqual(event.run_id, run.id) + self.assertEqual(event.artifact_id, "artifact-1") + self.assertEqual( + json.loads(event.source_json), {"uri": "https://example.com/source"} + ) + self.assertEqual(json.loads(event.locator_json), {"path": "/tmp/source"}) + self.assertEqual(json.loads(event.artifact_json), {"id": "artifact-1"}) + self.assertEqual(json.loads(event.harvest_json), {"status": "ok"}) + self.assertEqual( + event.observed_at.replace(tzinfo=None), + observed_at.astimezone(timezone.utc).replace(tzinfo=None), + ) + self.assertIsNotNone(event.created_at) + + chunk = db.create_ingest_chunk( + artifact_event_id=event.id, + chunk_id="chunk-1", + text="hello world", + char_count=11, + span_json={"start": 0, "end": 11}, + delta_json={"op": "add"}, + ) + + self.assertIsNotNone(chunk.id) + self.assertEqual(chunk.artifact_event_id, event.id) + self.assertEqual(chunk.chunk_id, "chunk-1") + self.assertEqual(chunk.text, "hello world") + self.assertEqual(chunk.char_count, 11) + self.assertEqual(json.loads(chunk.span_json), {"start": 0, "end": 11}) + self.assertEqual(json.loads(chunk.delta_json), {"op": "add"}) + self.assertIsNotNone(chunk.created_at) diff --git a/application/utils/harvester/change_detector.py b/application/utils/harvester/change_detector.py new file mode 100644 index 000000000..584d43d9c --- /dev/null +++ b/application/utils/harvester/change_detector.py @@ -0,0 +1,107 @@ +import logging +import subprocess + +from .git_repository_client import GitRepositoryClient + +logger = logging.getLogger(__name__) + + +class ChangeDetector: + def __init__(self, repository_client: GitRepositoryClient): + self.repository_client = repository_client + + def _resolve_commit(self, commit_sha: str) -> str: + try: + result = subprocess.run( + [ + "git", + "-C", + str(self.repository_client.get_local_path()), + "rev-parse", + "--verify", + "--end-of-options", + f"{commit_sha}^{{commit}}", + ], + capture_output=True, + text=True, + check=True, + timeout=60, + ) + except subprocess.CalledProcessError as exc: + logger.error("Git command failed: %s", exc.stderr) + raise + + return result.stdout.strip() + + def get_modified_files_since( + self, base_commit: str, target_commit: str + ) -> list[str]: + logger.info( + "Detecting changes between %s and %s", + base_commit, + target_commit, + ) + + base = self._resolve_commit(base_commit) + target = self._resolve_commit(target_commit) + + try: + result = subprocess.run( + [ + "git", + "-C", + str(self.repository_client.get_local_path()), + "diff", + "--name-only", + base, + target, + ], + capture_output=True, + text=True, + check=True, + timeout=60, + ) + except subprocess.CalledProcessError as exc: + logger.error("Git command failed: %s", exc.stderr) + raise + + files = [ + file_path for file_path in result.stdout.splitlines() if file_path.strip() + ] + + return sorted(set(files)) + + def get_commits_since(self, base_commit: str, target_commit: str) -> list[str]: + base = self._resolve_commit(base_commit) + target = self._resolve_commit(target_commit) + + try: + result = subprocess.run( + [ + "git", + "-C", + str(self.repository_client.get_local_path()), + "log", + "--reverse", + "--format=%H", + f"{base}..{target}", + ], + capture_output=True, + text=True, + check=True, + timeout=60, + ) + except subprocess.CalledProcessError as exc: + logger.error("Git command failed: %s", exc.stderr) + raise + + commits = [sha for sha in result.stdout.splitlines() if sha.strip()] + + logger.info( + "Detected %s commits between %s and %s", + len(commits), + base_commit, + target_commit, + ) + + return commits diff --git a/application/utils/harvester/checkpoint_store.py b/application/utils/harvester/checkpoint_store.py new file mode 100644 index 000000000..e9e3b62f4 --- /dev/null +++ b/application/utils/harvester/checkpoint_store.py @@ -0,0 +1,95 @@ +from typing import Any + +from sqlalchemy.exc import IntegrityError + +from application import sqla +from application.database.db import HarvesterCheckpoint +from .models import RepositoryCheckpoint + + +class CheckpointStore: + def __init__(self, session: Any = None) -> None: + self._session = session + + @property + def session(self) -> Any: + return self._session if self._session is not None else sqla.session + + def load(self, repository_id: str) -> RepositoryCheckpoint | None: + session = self.session + record = ( + session.query(HarvesterCheckpoint) + .filter_by(repository_id=repository_id) + .first() + ) + if record is None: + return None + return RepositoryCheckpoint( + repository_id=record.repository_id, + last_processed_commit=record.last_processed_commit, + updated_at=record.updated_at, + provider=record.provider, + owner=record.owner, + repository=record.repository, + branch=record.branch, + ) + + def save(self, checkpoint: RepositoryCheckpoint) -> None: + session = self.session + existing = ( + session.query(HarvesterCheckpoint) + .filter_by(repository_id=checkpoint.repository_id) + .first() + ) + + if existing is None: + canonical_conflict = ( + session.query(HarvesterCheckpoint) + .filter_by( + provider=checkpoint.provider, + owner=checkpoint.owner, + repository=checkpoint.repository, + branch=checkpoint.branch, + ) + .first() + ) + if canonical_conflict is not None: + session.rollback() + raise ValueError("duplicate canonical source identity") + + new_record = HarvesterCheckpoint( + repository_id=checkpoint.repository_id, + provider=checkpoint.provider, + owner=checkpoint.owner, + repository=checkpoint.repository, + branch=checkpoint.branch, + last_processed_commit=checkpoint.last_processed_commit, + updated_at=checkpoint.updated_at, + ) + session.add(new_record) + try: + session.commit() + except IntegrityError: + session.rollback() + raise ValueError("duplicate canonical source identity") + except Exception: + session.rollback() + raise + return + + if ( + existing.provider != checkpoint.provider + or existing.owner != checkpoint.owner + or existing.repository != checkpoint.repository + or existing.branch != checkpoint.branch + ): + session.rollback() + raise ValueError("immutable repository identity") + + existing.last_processed_commit = checkpoint.last_processed_commit + existing.updated_at = checkpoint.updated_at + try: + session.commit() + except Exception: + session.rollback() + raise diff --git a/application/utils/harvester/git_repository_client.py b/application/utils/harvester/git_repository_client.py index 1390e6606..bed925d85 100644 --- a/application/utils/harvester/git_repository_client.py +++ b/application/utils/harvester/git_repository_client.py @@ -20,6 +20,8 @@ def __init__( branch: str = "main", local_path: Path | None = None, ) -> None: + if branch.startswith("-"): + raise ValueError("Invalid git branch") self.owner = owner self.repository = repository self.branch = branch diff --git a/application/utils/harvester/models.py b/application/utils/harvester/models.py new file mode 100644 index 000000000..2050913c7 --- /dev/null +++ b/application/utils/harvester/models.py @@ -0,0 +1,20 @@ +from dataclasses import dataclass +from datetime import datetime + + +@dataclass(slots=True) +class RepositoryCheckpoint: + repository_id: str + last_processed_commit: str | None + updated_at: datetime + provider: str + owner: str + repository: str + branch: str + + +@dataclass(slots=True) +class RepositoryChangeSet: + repository_id: str + commit_sha: str + modified_files: list[str] diff --git a/application/utils/harvester/repository_cache.py b/application/utils/harvester/repository_cache.py index 94b721a60..a8c9b632c 100644 --- a/application/utils/harvester/repository_cache.py +++ b/application/utils/harvester/repository_cache.py @@ -18,6 +18,9 @@ def build_repository_cache_path( if not _VALID_COMPONENT.fullmatch(repository): raise ValueError(f"Invalid repository name: {repository}") + if branch in {".", ".."}: + raise ValueError("Invalid branch name") + encoded_branch = quote(branch, safe="") candidate = CACHE_ROOT / owner.casefold() / repository.casefold() / encoded_branch diff --git a/application/utils/harvester/repository_lock.py b/application/utils/harvester/repository_lock.py index 9890e6e2e..a9779033b 100644 --- a/application/utils/harvester/repository_lock.py +++ b/application/utils/harvester/repository_lock.py @@ -14,7 +14,7 @@ def repository_lock(repository_path: Path): Acquire an exclusive inter-process lock for a repository cache path. """ - lock_path = repository_path.with_suffix(".lock") + lock_path = repository_path.parent / f"{repository_path.name}.lock" lock_path.parent.mkdir(parents=True, exist_ok=True) with lock_path.open("w") as lock_file: diff --git a/migrations/versions/6a9d0d62ef41_add_harvester_checkpoint_table.py b/migrations/versions/6a9d0d62ef41_add_harvester_checkpoint_table.py new file mode 100644 index 000000000..4a3ed2d54 --- /dev/null +++ b/migrations/versions/6a9d0d62ef41_add_harvester_checkpoint_table.py @@ -0,0 +1,41 @@ +"""add harvester_checkpoint table + +Revision ID: 6a9d0d62ef41 +Revises: 9f1a2b3c4d5e +Create Date: 2026-07-25 + +""" + +from alembic import op +import sqlalchemy as sa + + +revision = "6a9d0d62ef41" +down_revision = "9f1a2b3c4d5e" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "harvester_checkpoint", + sa.Column("repository_id", sa.String(), primary_key=True), + sa.Column("provider", sa.String(), nullable=False), + sa.Column("owner", sa.String(), nullable=False), + sa.Column("repository", sa.String(), nullable=False), + sa.Column("branch", sa.String(), nullable=False), + sa.Column("last_processed_commit", sa.String(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.UniqueConstraint( + "provider", + "owner", + "repository", + "branch", + name="uq_harvester_checkpoint_canonical_source", + ), + ) + + +def downgrade(): + op.drop_table("harvester_checkpoint") diff --git a/migrations/versions/9f1a2b3c4d5e_add_artifact_ingest_persistence.py b/migrations/versions/9f1a2b3c4d5e_add_artifact_ingest_persistence.py new file mode 100644 index 000000000..9fad7bed6 --- /dev/null +++ b/migrations/versions/9f1a2b3c4d5e_add_artifact_ingest_persistence.py @@ -0,0 +1,82 @@ +"""add artifact ingest event and chunk tables + +Revision ID: 9f1a2b3c4d5e +Revises: e1f2a3b4c5d6 +Create Date: 2026-07-23 + +""" + +from alembic import op +import sqlalchemy as sa + + +revision = "9f1a2b3c4d5e" +down_revision = "c7d8e9f0a1b2" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "artifact_ingest_event", + sa.Column("id", sa.String(), primary_key=True), + sa.Column("run_id", sa.String(), nullable=False), + sa.Column("artifact_id", sa.String(), nullable=False), + sa.Column("harvest_mode", sa.String(), nullable=False), + sa.Column("event_type", sa.String(), nullable=False), + sa.Column("source_json", sa.Text(), nullable=False), + sa.Column("locator_json", sa.Text(), nullable=False), + sa.Column("artifact_json", sa.Text(), nullable=False), + sa.Column("harvest_json", sa.Text(), nullable=False), + sa.Column("observed_at", sa.DateTime(), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint( + ["run_id"], + ["import_run.id"], + onupdate="CASCADE", + ondelete="CASCADE", + ), + ) + op.create_unique_constraint( + "uq_artifact_ingest_event_run_artifact", + "artifact_ingest_event", + ["run_id", "artifact_id"], + ) + + op.create_table( + "ingest_chunk", + sa.Column("id", sa.String(), primary_key=True), + sa.Column("artifact_event_id", sa.String(), nullable=False), + sa.Column("chunk_id", sa.String(), nullable=False), + sa.Column("text", sa.Text(), nullable=False), + sa.Column("char_count", sa.Integer(), nullable=False), + sa.Column("span_json", sa.Text(), nullable=False), + sa.Column("delta_json", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint( + ["artifact_event_id"], + ["artifact_ingest_event.id"], + onupdate="CASCADE", + ondelete="CASCADE", + ), + ) + op.create_unique_constraint( + "uq_ingest_chunk_artifact_chunk", + "ingest_chunk", + ["artifact_event_id", "chunk_id"], + ) + + +def downgrade(): + op.drop_constraint( + "uq_ingest_chunk_artifact_chunk", + "ingest_chunk", + type_="unique", + ) + op.drop_table("ingest_chunk") + op.drop_constraint( + "uq_artifact_ingest_event_run_artifact", + "artifact_ingest_event", + type_="unique", + ) + op.drop_table("artifact_ingest_event")