diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..a587e5f --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,32 @@ +name: CI + +on: + push: + pull_request: + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - name: Install dev dependencies + run: pip install -e .[dev] + - name: Lint with ruff + run: ruff check . + - name: Check formatting with black + run: black --check . + + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - name: Install dev dependencies + run: pip install -e .[dev] + - name: Run tests + run: pytest diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c1b2887 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ +.pytest_cache/ +.ruff_cache/ +*.egg diff --git a/CLAUDE.md b/CLAUDE.md index 1318c33..3d3ab25 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,11 +47,17 @@ Each package has a single responsibility. Do not let reconciliation logic bleed ## Tests -- Framework: `pytest` +**Every new module must ship with tests. A feature without a test file is not done.** + +Full guide: [docs/testing.md](docs/testing.md). + +- Framework: `pytest` — run with `pytest` from the repo root. +- Test files mirror the source tree: `src/sync_engine/store/base.py` → `tests/store/test_*.py`. - One test per behavior, not per function. -- Webhook tests mock signature verification (do not depend on secrets in tests). -- Reconciliation tests mock the REST API (no network calls in CI). - Naming: `test___`. +- Webhook tests mock signature verification (never depend on secrets). +- Reconciliation tests mock the REST API (no network calls in CI). +- Use `InMemoryStore` as the store fixture — never a real backend. --- @@ -63,6 +69,29 @@ Each package has a single responsibility. Do not let reconciliation logic bleed - Valid statuses: `Proposed` | `Accepted` | `Deprecated` | `Superseded by ADR-NNN` - An ADR is not modified after acceptance — create a superseding ADR instead. +### When to write a new ADR + +After every implementation session, ask: *did this introduce an architectural decision not covered by an existing ADR?* + +Write a new ADR when the decision: +- Chooses between two or more structurally different approaches (e.g. abstract interface vs. concrete dependency, REST vs. event-driven). +- Has cross-cutting consequences — it constrains how multiple packages or future implementations must behave. +- Would be non-obvious to reverse without touching many files. + +Do **not** write an ADR for: +- Code conventions already documented in this file (exception hierarchy, naming, formatting). +- Single-file implementation choices with no downstream constraints. +- Bug fixes or adjustments to existing decisions. + +If a new implementation *contradicts* an existing ADR, the new ADR must reference it with status `Superseded by ADR-NNN`. + +### Current ADR index + +| ADR | Title | Status | +|---|---|---| +| [ADR-001](docs/adr/ADR-001-hybrid-rest-webhook.md) | Hybrid REST/webhook architecture | Accepted | +| [ADR-002](docs/adr/ADR-002-pluggable-store-interface.md) | Pluggable store interface via abstract base class | Accepted | + --- ## Git @@ -71,3 +100,13 @@ Each package has a single responsibility. Do not let reconciliation logic bleed - Commit messages in English, imperative mood, no trailing period. - Format: `: ` — types: `feat`, `fix`, `docs`, `refactor`, `test`, `chore` - Do not commit secrets, `.env` files, or tokens. + +--- + +## PR review workflow + +After pushing a PR branch: + +1. Read all open review threads (`get_review_comments`). +2. For each thread: apply the fix, reply with the commit SHA and a one-line explanation, then resolve the thread. +3. Once all threads are resolved, post `@codex review` as a top-level PR comment to trigger a follow-up automated review. diff --git a/README.md b/README.md index 5158318..6b402dc 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ # sync-engine +[![CI](https://github.com/koydas/sync-engine/actions/workflows/ci.yml/badge.svg)](https://github.com/koydas/sync-engine/actions/workflows/ci.yml) + Hybrid REST/webhook sync engine — push/pull coordination, idempotent processing, and failure recovery. Extracted from production integration patterns of `fullstack-pilot`. --- diff --git a/docs/adr/ADR-002-pluggable-store-interface.md b/docs/adr/ADR-002-pluggable-store-interface.md new file mode 100644 index 0000000..8e01b58 --- /dev/null +++ b/docs/adr/ADR-002-pluggable-store-interface.md @@ -0,0 +1,69 @@ +# ADR-002 — Pluggable store interface via abstract base class + +**Status**: Accepted +**Date**: 2026-06-09 +**Deciders**: fullstack-pilot team + +--- + +## Context + +ADR-001 establishes that a persistent store is required for two things: + +1. `last_successful_sync_at` — used by the reconciliation loop to compute the missed-event window after an outage. +2. The webhook queue — at-least-once delivery buffer that survives a crash between reception and processing. + +ADR-001 does not specify the implementation technology or how the rest of the engine should depend on the store. Two options were considered: + +1. **Concrete dependency** — engine components import and instantiate a specific backend directly (e.g. SQLite helper, Redis client). +2. **Abstract interface** — engine components depend on a Python ABC; the concrete backend is injected at startup. + +--- + +## Decision + +We define `SyncStore` as a Python abstract base class (`abc.ABC`) in `src/sync_engine/store/base.py`. All engine components — webhook handler, reconciliation loop, processor — interact exclusively with this interface. No engine code imports a concrete backend. + +The interface surface is intentionally minimal (six methods): + +| Method | Purpose | +|---|---| +| `get_last_sync_at()` | Read the recovery timestamp | +| `set_last_sync_at(dt)` | Write it after a confirmed commit | +| `is_event_processed(event_id)` | Idempotency check before processing | +| `mark_event_processed(event_id)` | Acknowledge after commit | +| `enqueue_webhook(event_id, payload)` | Persist before processing begins | +| `dequeue_unacknowledged()` | Replay buffer on recovery | + +`dequeue_unacknowledged` returns `list[tuple[str, dict]]` — event_id is always returned alongside the payload so callers can acknowledge without the payload embedding its own ID. + +--- + +## Why not a concrete dependency + +- **Test coupling**: any concrete backend (SQLite, Redis) requires setup, teardown, and I/O in tests. An in-memory implementation behind the same interface eliminates all of that. +- **Migration cost**: coupling engine logic to a specific backend turns a backend swap into a cross-cutting refactor. With an interface, only a new implementation file is needed. +- **Invariant enforcement**: the interface is the right place to document the commit-before-acknowledge invariants (see CLAUDE.md §Invariants). Concrete classes are not readable enough for this. + +--- + +## Consequences + +**Positive:** +- `InMemoryStore` (`store/memory.py`) satisfies the interface for all tests — no I/O, no fixtures, no external process. +- Production backends (SQLite, Redis, Postgres) are drop-in replacements — they implement the ABC and are injected at startup. +- Engine invariants (acknowledge-after-commit, last_sync_at write-only-after-success) are enforced at the call site by convention documented on the interface, not scattered across backends. + +**Negative / accepted costs:** +- Dependency injection must be wired at startup — the concrete backend is not self-instantiated. +- Adding a new persistence operation requires updating the ABC and every implementation, not just one file. + +--- + +## Rejected alternatives + +| Alternative | Reason for rejection | +|---|---| +| Hardcoded SQLite backend | Tight coupling; tests require a real file; migration is cross-cutting | +| Dataclass / TypedDict (no ABC) | No enforcement of the interface contract; silent drift between implementations | +| Repository pattern with separate read/write interfaces | Over-engineering for current scope; a single interface covers all six operations cleanly | diff --git a/docs/testing.md b/docs/testing.md new file mode 100644 index 0000000..bf18c3b --- /dev/null +++ b/docs/testing.md @@ -0,0 +1,119 @@ +# Testing guide + +## Running tests + +```bash +pip install -e ".[dev]" +pytest # all tests +pytest -v # verbose +pytest tests/store/ # one subdirectory +``` + +--- + +## Layout + +Tests mirror the source tree under `tests/`: + +``` +tests/ +├── test_exceptions.py +├── store/ +│ └── test_memory_store.py +├── webhook/ +│ └── test_handler.py # (upcoming) +├── reconciliation/ +│ └── test_loop.py # (upcoming) +└── processor/ + └── test_processor.py # (upcoming) +``` + +Every new module in `src/sync_engine//` gets a matching test file +in `tests//`. No test file = the feature is not done. + +--- + +## Naming + +``` +test___ +``` + +One test per behavior, not per function. Examples: + +```python +def test_get_last_sync_at_before_any_write_returns_none(): ... +def test_dequeue_unacknowledged_excludes_processed_events(): ... +def test_webhook_signature_error_caught_as_sync_error(): ... +``` + +--- + +## Fixtures + +Define fixtures at the test-file level unless shared across multiple files, in +which case add them to a `conftest.py` in the relevant directory. + +`InMemoryStore` is the canonical fixture for any test that touches the store: + +```python +import pytest +from sync_engine.store.memory import InMemoryStore + +@pytest.fixture +def store() -> InMemoryStore: + return InMemoryStore() +``` + +--- + +## Mocking rules + +### Webhook signature verification + +Never depend on real secrets in tests. Patch the verification call: + +```python +from unittest.mock import patch + +def test_valid_webhook_is_queued(store): + with patch("sync_engine.webhook.handler.verify_signature", return_value=True): + ... +``` + +### REST API (reconciliation) + +No network calls in CI. Patch `httpx.Client` or inject a fake transport: + +```python +import httpx + +def test_reconciliation_fetches_since_last_sync(store): + def handler(request): + return httpx.Response(200, json={"items": []}) + + transport = httpx.MockTransport(handler) + client = httpx.Client(transport=transport) + ... +``` + +--- + +## What to test + +For each new module, cover at minimum: + +| Scenario | Example | +|---|---| +| Happy path | event processed end-to-end | +| Idempotency | same `event_id` processed twice, second is a no-op | +| Pre-condition violated | invalid signature → `WebhookSignatureError` raised | +| State after failure | crash mid-processing → event still in `dequeue_unacknowledged` | + +--- + +## What not to test + +- Internal implementation details (private methods, internal dict structure). +- The abstract `SyncStore` interface directly — test it through `InMemoryStore`. +- Framework behavior (FastAPI routing, httpx serialization). diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..2e6679d --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,26 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "sync-engine" +version = "0.1.0" +requires-python = ">=3.11" +dependencies = [ + "fastapi", + "httpx", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "ruff", + "black", +] + +[tool.hatch.build.targets.wheel] +packages = ["src/sync_engine"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["src"] diff --git a/src/sync_engine/__init__.py b/src/sync_engine/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/sync_engine/exceptions.py b/src/sync_engine/exceptions.py new file mode 100644 index 0000000..a0fe76e --- /dev/null +++ b/src/sync_engine/exceptions.py @@ -0,0 +1,14 @@ +class SyncError(Exception): + """Base class for all sync-engine business exceptions.""" + + +class WebhookSignatureError(SyncError): + """Raised when a webhook payload fails HMAC signature verification.""" + + +class DuplicateEventError(SyncError): + """Raised when an event_id has already been processed (idempotency guard).""" + + +class ReconciliationError(SyncError): + """Raised when the REST reconciliation loop encounters an unrecoverable error.""" diff --git a/src/sync_engine/store/__init__.py b/src/sync_engine/store/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/sync_engine/store/base.py b/src/sync_engine/store/base.py new file mode 100644 index 0000000..8abf788 --- /dev/null +++ b/src/sync_engine/store/base.py @@ -0,0 +1,32 @@ +from abc import ABC, abstractmethod +from datetime import datetime + + +class SyncStore(ABC): + @abstractmethod + def get_last_sync_at(self) -> datetime | None: + """Return the timestamp of the last confirmed successful sync, or None.""" + + @abstractmethod + def set_last_sync_at(self, dt: datetime) -> None: + """Persist dt as the last successful sync timestamp. + + Must only be called after the sync has been fully committed — never + at the start of a cycle (see invariant #3 in CLAUDE.md). + """ + + @abstractmethod + def is_event_processed(self, event_id: str) -> bool: + """Return True if event_id has already been committed (idempotency check).""" + + @abstractmethod + def mark_event_processed(self, event_id: str) -> None: + """Record event_id as processed. Call only after the payload is committed.""" + + @abstractmethod + def enqueue_webhook(self, event_id: str, payload: dict) -> None: + """Persist a webhook payload before processing begins (at-least-once delivery).""" + + @abstractmethod + def dequeue_unacknowledged(self) -> list[tuple[str, dict]]: + """Return (event_id, payload) pairs for every queued webhook not yet marked processed.""" diff --git a/src/sync_engine/store/memory.py b/src/sync_engine/store/memory.py new file mode 100644 index 0000000..e745638 --- /dev/null +++ b/src/sync_engine/store/memory.py @@ -0,0 +1,34 @@ +from datetime import datetime + +from sync_engine.store.base import SyncStore + + +class InMemoryStore(SyncStore): + """In-memory SyncStore implementation — for tests only, no persistence.""" + + def __init__(self) -> None: + self._last_sync_at: datetime | None = None + self._processed: set[str] = set() + self._queue: dict[str, dict] = {} + + def get_last_sync_at(self) -> datetime | None: + return self._last_sync_at + + def set_last_sync_at(self, dt: datetime) -> None: + self._last_sync_at = dt + + def is_event_processed(self, event_id: str) -> bool: + return event_id in self._processed + + def mark_event_processed(self, event_id: str) -> None: + self._processed.add(event_id) + + def enqueue_webhook(self, event_id: str, payload: dict) -> None: + self._queue[event_id] = payload + + def dequeue_unacknowledged(self) -> list[tuple[str, dict]]: + return [ + (event_id, payload) + for event_id, payload in self._queue.items() + if event_id not in self._processed + ] diff --git a/tests/store/test_memory_store.py b/tests/store/test_memory_store.py new file mode 100644 index 0000000..4cf385a --- /dev/null +++ b/tests/store/test_memory_store.py @@ -0,0 +1,74 @@ +from datetime import datetime, timezone + +import pytest + +from sync_engine.store.memory import InMemoryStore + + +@pytest.fixture +def store() -> InMemoryStore: + return InMemoryStore() + + +def test_get_last_sync_at_before_any_write_returns_none(store): + assert store.get_last_sync_at() is None + + +def test_set_last_sync_at_get_returns_same_value(store): + dt = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + store.set_last_sync_at(dt) + assert store.get_last_sync_at() == dt + + +def test_set_last_sync_at_overwrites_previous_value(store): + first = datetime(2026, 1, 1, tzinfo=timezone.utc) + second = datetime(2026, 6, 1, tzinfo=timezone.utc) + store.set_last_sync_at(first) + store.set_last_sync_at(second) + assert store.get_last_sync_at() == second + + +def test_is_event_processed_unknown_event_returns_false(store): + assert store.is_event_processed("evt-123") is False + + +def test_is_event_processed_after_mark_returns_true(store): + store.mark_event_processed("evt-123") + assert store.is_event_processed("evt-123") is True + + +def test_mark_event_processed_called_twice_is_idempotent(store): + store.mark_event_processed("evt-123") + store.mark_event_processed("evt-123") + assert store.is_event_processed("evt-123") is True + + +def test_enqueue_webhook_appears_in_dequeue_unacknowledged(store): + store.enqueue_webhook("evt-1", {"type": "created"}) + assert ("evt-1", {"type": "created"}) in store.dequeue_unacknowledged() + + +def test_dequeue_unacknowledged_returns_event_id_alongside_payload(store): + store.enqueue_webhook("evt-1", {"type": "created"}) + event_id, payload = store.dequeue_unacknowledged()[0] + assert event_id == "evt-1" + assert payload == {"type": "created"} + + +def test_dequeue_unacknowledged_empty_before_any_enqueue(store): + assert store.dequeue_unacknowledged() == [] + + +def test_dequeue_unacknowledged_excludes_processed_events(store): + store.enqueue_webhook("evt-1", {"type": "created"}) + store.mark_event_processed("evt-1") + assert store.dequeue_unacknowledged() == [] + + +def test_dequeue_unacknowledged_returns_only_unprocessed_from_mixed_queue(store): + store.enqueue_webhook("evt-1", {"type": "created"}) + store.enqueue_webhook("evt-2", {"type": "updated"}) + store.mark_event_processed("evt-1") + event_ids = [eid for eid, _ in store.dequeue_unacknowledged()] + assert "evt-1" not in event_ids + assert "evt-2" in event_ids diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py new file mode 100644 index 0000000..6746315 --- /dev/null +++ b/tests/test_exceptions.py @@ -0,0 +1,28 @@ +import pytest + +from sync_engine.exceptions import ( + DuplicateEventError, + ReconciliationError, + SyncError, + WebhookSignatureError, +) + + +def test_webhook_signature_error_caught_as_sync_error(): + with pytest.raises(SyncError): + raise WebhookSignatureError("invalid signature") + + +def test_duplicate_event_error_caught_as_sync_error(): + with pytest.raises(SyncError): + raise DuplicateEventError("evt-123 already processed") + + +def test_reconciliation_error_caught_as_sync_error(): + with pytest.raises(SyncError): + raise ReconciliationError("REST call failed") + + +def test_sync_error_caught_as_base_exception(): + with pytest.raises(Exception): + raise SyncError("base error")