Chris/schy 45 datastreamreplicator for python

.claude/skills/pr-review/SKILL.md

@@ -0,0 +1,129 @@
+---
+name: pr-review
+description: Review code changes on the current branch for quality, bugs, performance, and security
+disable-model-invocation: true
+argument-hint: "[optional: LINEAR-TICKET-ID]"
+allowed-tools: Read, Grep, Glob, Bash(git diff:*), Bash(git log:*), Bash(git show:*), Bash(git branch:*), Bash(gh pr:*), Bash(gh api:*), Bash(~/.claude/scripts/fetch-github-pr.sh:*), Bash(~/.claude/scripts/fetch-sentry-data.sh:*), Bash(~/.claude/scripts/fetch-slack-thread.sh:*)
+---
+
+# Code Review
+
+You are reviewing code changes on the current branch. Your review must be based on the **current state of the code right now**, not on anything you've seen earlier in this conversation.
+
+## CRITICAL: Always Use Fresh Data
+
+**IGNORE any file contents, diffs, or line numbers you may have seen earlier in this conversation.** They may be stale. You MUST re-fetch everything from scratch using the commands below.
+
+## Step 1: Get the Current Diff and PR Context
+
+Run ALL of these commands to get a fresh view:
+
+```bash
+# The authoritative diff -- only review what's in HERE
+git diff main...HEAD
+
+# Recent commits on this branch
+git log --oneline main..HEAD
+
+# PR description and comments
+gh pr view --json number,title,body,comments,reviews,reviewRequests
+```
+
+Also fetch PR review comments (inline code comments):
+
+```bash
+# Get the PR number
+PR_NUMBER=$(gh pr view --json number -q '.number')
+
+# Fetch all review comments (inline comments on specific lines)
+gh api repos/{owner}/{repo}/pulls/$PR_NUMBER/comments --jq '.[] | {path: .path, line: .line, body: .body, user: .user.login, created_at: .created_at}'
+
+# Fetch review-level comments (general review comments)
+gh api repos/{owner}/{repo}/pulls/$PR_NUMBER/reviews --jq '.[] | {state: .state, body: .body, user: .user.login}'
+```
+
+## Step 2: Understand Context from PR Comments
+
+Before reviewing, read through the PR comments and review comments. Note **who** said what (by username).
+
+- **Already-addressed feedback**: If a reviewer pointed out an issue and the author has already fixed it (the fix is visible in the current diff), do NOT re-raise it.
+- **Ongoing discussions**: Note any unresolved threads -- your review should take these into account.
+- **Previous approvals/requests for changes**: Understand what reviewers have already looked at.
+
+**IMPORTANT**: Your review is YOUR independent review. Do not take credit for or reference other reviewers' findings as if they were yours. If another reviewer already flagged something, you can note "as [reviewer] pointed out" but do not present their feedback as your own prior review. Your verdict should be based solely on your own analysis of the current code.
+
+## Step 3: Get Requirements Context
+
+Check if a Linear ticket ID was provided as an argument ($ARGUMENTS). If not, try to extract it from the branch name (pattern: `{username}/{linear-ticket}-{title}`).
+
+If a Linear ticket is found:
+- Use Linear MCP tools (`get_issue`) to get the issue details and comments
+- **Check for a parent ticket**: If the issue has a parent issue, fetch the parent too. Our pattern is to have a parent ticket with project-wide requirements and sub-tickets for specific tasks (often one per repo/PR). The parent ticket will contain the full scope of the project, while the sub-ticket scopes what this specific PR should cover. Use both to assess completeness — the PR should fulfill the sub-ticket's scope, and that scope should be a reasonable subset of the parent's backend-related requirements.
+- Look for Sentry links in the description/comments; if found, use Sentry MCP tools to get error details
+- Assess whether the changes fulfill the ticket requirements
+
+If no ticket is found, check the PR description for context on what the changes are meant to accomplish.
+
+## Step 4: Review the Code
+
+Review ONLY the changed lines (from `git diff main...HEAD`). Do not comment on unchanged code.
+
+**When referencing code, always use the file path and quote the actual code snippet** rather than citing line numbers, since line numbers shift as the branch evolves.
+
+### Code Quality
+- Is the code well-structured and maintainable?
+- Does it follow CLAUDE.md conventions? (import grouping, error handling with lib/errors, naming, alphabetization, etc.)
+- Any AI-generated slop? (excessive comments, unnecessary abstractions, over-engineering)
+
+### Performance
+- N+1 queries, inefficient loops, missing indexes for new queries
+- Unbuffered writes in hot paths (especially ClickHouse)
+- Missing LIMIT clauses on potentially large result sets
+
+### Bugs
+- Nil pointer risks (especially on struct pointer params and optional relations)
+- Functions returning `nil, nil` (violates convention)
+- Missing error handling
+- Race conditions in concurrent code paths
+
+### Security
+- Hardcoded secrets or sensitive data exposure
+- Missing input validation on service request structs
+
+### Tests
+- Are there tests for the new/changed code?
+- Do the tests cover edge cases and error paths?
+- Are test assertions specific (not just "no error")?
+
+## Step 5: Present the Review
+
+Structure your review as:
+
+```
+## Summary
+[1-2 sentences: what this PR does and overall assessment]
+
+## Requirements Check
+[Does the PR fulfill the Linear ticket / PR description requirements? Any gaps?]
+
+## Issues
+### Critical (must fix before merge)
+- [blocking issues]
+
+### Suggestions (nice to have)
+- [non-blocking improvements]
+
+## Prior Review Activity
+[Summarize what other reviewers have flagged, attributed by name. Note which of their concerns have been addressed in the current code and which remain open.]
+
+## Verdict
+[LGTM / Needs changes / Needs discussion -- based on YOUR analysis, not other reviewers' findings]
+```
+
+## Guidelines
+
+- Be concise. Don't pad with praise or filler.
+- Only raise issues that matter. Don't nitpick formatting (that's what linters are for).
+- Quote code snippets rather than referencing line numbers.
+- If PR comments show a discussion was already resolved, don't reopen it.
+- If you're unsure about something, flag it as a question rather than a definitive issue.

.fernignore

@@ -7,14 +7,19 @@ README.md
 src/schematic/client.py
 
 # Additional custom code
+.claude/
 .github/CODEOWNERS
+.github/workflows/ci.yml
 scripts/
-src/schematic/cache.py
+src/schematic/cache/
 src/schematic/event_buffer.py
 src/schematic/http_client.py
 src/schematic/logging.py
+src/schematic/datastream/
 src/schematic/webhook_utils/
 src/schematic/webhooks/verification.py
 tests/custom/
+tests/datastream/
 tests/webhook_utils/
 CLAUDE.md
+WASM_VERSION

.github/workflows/ci.yml

@@ -1,5 +1,7 @@
 name: ci
 on: [push]
+env:
+  GH_TOKEN: ${{ secrets.GH_TOKEN }}
 jobs:
   compile:
     runs-on: ubuntu-latest
@@ -13,8 +15,10 @@ jobs:
       - name: Bootstrap poetry
         run: |
           curl -sSL https://install.python-poetry.org | python - -y --version 1.5.1
+      - name: Download WASM binary
+        run: ./scripts/download-wasm.sh
       - name: Install dependencies
-        run: poetry install
+        run: poetry install --extras datastream
       - name: Compile
         run: poetry run mypy .
   test:
@@ -29,14 +33,42 @@ jobs:
       - name: Bootstrap poetry
         run: |
           curl -sSL https://install.python-poetry.org | python - -y --version 1.5.1
+      - name: Download WASM binary
+        run: ./scripts/download-wasm.sh
       - name: Install dependencies
-        run: poetry install
+        run: poetry install --extras datastream
 
       - name: Test
         run: poetry run pytest -rP -n auto .
 
+  verify-package:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repo
+        uses: actions/checkout@v4
+      - name: Set up python
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.9
+      - name: Bootstrap poetry
+        run: |
+          curl -sSL https://install.python-poetry.org | python - -y --version 1.5.1
+      - name: Download WASM binary
+        run: ./scripts/download-wasm.sh
+      - name: Build package
+        run: poetry build
+      - name: Verify WASM in wheel
+        run: |
+          if ! zipinfo dist/*.whl | grep -q 'rulesengine.wasm'; then
+            echo "ERROR: rulesengine.wasm not found in wheel"
+            echo "Wheel contents:"
+            zipinfo dist/*.whl
+            exit 1
+          fi
+          echo "Verified: rulesengine.wasm is included in the wheel"
+
   publish:
-    needs: [compile, test]
+    needs: [compile, test, verify-package]
     if: github.event_name == 'push' && contains(github.ref, 'refs/tags/')
     runs-on: ubuntu-latest
     steps:
@@ -49,6 +81,8 @@ jobs:
       - name: Bootstrap poetry
         run: |
           curl -sSL https://install.python-poetry.org | python - -y --version 1.5.1
+      - name: Download WASM binary
+        run: ./scripts/download-wasm.sh
       - name: Install dependencies
         run: poetry install
       - name: Publish to pypi

.gitignore

@@ -3,3 +3,7 @@
 __pycache__/
 dist/
 poetry.toml
+
+# WASM binary (downloaded from rulesengine-rust GitHub Releases)
+src/schematic/datastream/wasm/rulesengine.wasm
+src/schematic/datastream/wasm/.wasm_version

README.md

@@ -285,6 +285,129 @@ client.check_flag(
 )
 ```
 
+## DataStream
+
+The DataStream functionality provides real-time updates for flags, companies, and users. It uses WebSocket connections to receive updates from the Schematic backend and evaluates feature flags locally using a WASM rules engine, reducing the number of network calls.
+
+### Installation
+
+DataStream requires additional dependencies for WebSocket connections and local flag evaluation. Install them with the `datastream` extra:
+
+```bash
+pip install 'schematichq[datastream]'
+# or
+poetry add schematichq -E datastream
+```
+
+### Key Features
+
+- **Real-Time Updates**: Automatically updates cached data when changes occur on the backend.
+- **Local Flag Evaluation**: Flag checks are evaluated locally via WASM, eliminating per-check network requests.
+- **Configurable Caching**: Supports both in-memory caching and custom cache providers (e.g. Redis).
+
+### How to Enable DataStream
+
+To enable DataStream, set `use_datastream=True` in your `AsyncSchematicConfig`:
+
+```python
+import asyncio
+from schematic.client import AsyncSchematic, AsyncSchematicConfig, DataStreamConfig
+
+async def main():
+    config = AsyncSchematicConfig(
+        use_datastream=True,
+    )
+
+    async with AsyncSchematic("YOUR_API_KEY", config) as client:
+        is_enabled = await client.check_flag(
+            "some-flag-key",
+            company={"id": "your-company-id"},
+            user={"id": "your-user-id"},
+        )
+
+asyncio.run(main())
+```
+
+### Configuring Cache TTL
+
+You can customize the cache TTL (in milliseconds) via the `DataStreamConfig`:
+
+```python
+config = AsyncSchematicConfig(
+    use_datastream=True,
+    datastream=DataStreamConfig(
+        cache_ttl=300_000,  # 5 minutes
+    ),
+)
+```
+
+### Replicator Mode
+
+When running the `schematic-datastream-replicator` service, configure the client to operate in Replicator Mode. In this mode, the client skips establishing its own WebSocket connection and instead relies on a shared cache populated by the external replicator service.
+
+```python
+import asyncio
+from schematic.client import AsyncSchematic, AsyncSchematicConfig, DataStreamConfig
+
+async def main():
+    config = AsyncSchematicConfig(
+        use_datastream=True,
+        datastream=DataStreamConfig(
+            replicator_mode=True,
+        ),
+    )
+
+    async with AsyncSchematic("YOUR_API_KEY", config) as client:
+        is_enabled = await client.check_flag(
+            "some-flag-key",
+            company={"id": "your-company-id"},
+        )
+
+asyncio.run(main())
+```
+
+#### Cache TTL Configuration
+
+When using Replicator Mode, you should set the SDK's cache TTL to match the replicator's cache TTL. The replicator defaults to an unlimited cache TTL. If the SDK uses a shorter TTL (the default is 24 hours), locally updated cache entries will be written back with the shorter TTL and eventually evicted from the shared cache.
+
+To match the replicator's default unlimited TTL:
+
+```python
+config = AsyncSchematicConfig(
+    use_datastream=True,
+    datastream=DataStreamConfig(
+        replicator_mode=True,
+        cache_ttl=None,  # Unlimited, matching the replicator default
+    ),
+)
+```
+
+#### Advanced Configuration
+
+```python
+config = AsyncSchematicConfig(
+    use_datastream=True,
+    datastream=DataStreamConfig(
+        replicator_mode=True,
+        cache_ttl=None,
+        replicator_health_url="http://my-replicator:8090/ready",
+        replicator_health_check=60_000,  # 60 seconds, in milliseconds
+    ),
+)
+```
+
+#### Default Configuration
+
+- **Replicator Health URL**: `http://localhost:8090/ready`
+- **Health Check Interval**: 30 seconds
+- **Cache TTL**: 24 hours (SDK default; should be set to match the replicator's TTL, which defaults to unlimited)
+
+When running in Replicator Mode, the client will:
+- Skip establishing WebSocket connections
+- Periodically check if the replicator service is ready
+- Use cached data populated by the external replicator service
+- Fall back to direct API calls if the replicator is not available
+
 ## Webhook Verification
 
 Schematic can send webhooks to notify your application of events. To ensure the security of these webhooks, Schematic signs each request using HMAC-SHA256. The Python SDK provides utility functions to verify these signatures.

WASM_VERSION

@@ -0,0 +1 @@
+0.1.0