Chris/schy 46 datastreamreplicator for java

.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

@@ -1,16 +1,24 @@
 # Specify files that shouldn't be modified by Fern
 CLAUDE.md
+.claude/
 LICENSE
 README.md
 .github/CODEOWNERS
+.github/workflows/ci.yml
+WASM_VERSION
+scripts/
 src/main/java/com/schematic/api/BaseSchematic.java
 src/main/java/com/schematic/api/EventBuffer.java
+src/main/java/com/schematic/api/HttpEventSender.java
 src/main/java/com/schematic/api/Schematic.java
 src/main/java/com/schematic/api/cache/CacheProvider.java
 src/main/java/com/schematic/api/cache/CachedItem.java
 src/main/java/com/schematic/api/cache/LocalCache.java
+src/main/java/com/schematic/api/cache/RedisCacheConfig.java
+src/main/java/com/schematic/api/cache/RedisCacheProvider.java
 src/main/java/com/schematic/api/core/NoOpHttpClient.java
 src/main/java/com/schematic/api/logger/ConsoleLogger.java
+src/main/java/com/schematic/api/datastream/
 src/main/java/com/schematic/api/logger/SchematicLogger.java
 src/main/java/com/schematic/webhook/
 src/test/java/com/schematic/api/TestCache.java
@@ -19,4 +27,5 @@ src/test/java/com/schematic/api/TestLogger.java
 src/test/java/com/schematic/api/TestOfflineMode.java
 src/test/java/com/schematic/api/TestReadme.java
 src/test/java/com/schematic/api/TestSchematic.java
+src/test/java/com/schematic/api/datastream/
 src/test/java/com/schematic/webhook/

.github/workflows/ci.yml

@@ -2,6 +2,9 @@ name: ci
 
 on: [push]
 
+env:
+  GH_TOKEN: ${{ secrets.GH_TOKEN }}
+
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
   cancel-in-progress: false
@@ -38,8 +41,12 @@ jobs:
           java-version: "11"
           architecture: x64
 
+      - name: Download WASM binary
+        run: ./scripts/download-wasm.sh
+
       - name: Test
         run: ./gradlew test
+
   publish:
     needs: [ compile, test ]
     if: github.event_name == 'push' && contains(github.ref, 'refs/tags/')
@@ -56,6 +63,9 @@ jobs:
           java-version: "11"
           architecture: x64
 
+      - name: Download WASM binary
+        run: ./scripts/download-wasm.sh
+
       - name: Publish to maven
         run: |
           ./gradlew sonatypeCentralUpload

.gitignore

@@ -21,4 +21,7 @@ generated_testSrc/
 generated/
 
 bin
-build
+build
+
+# WASM binary (downloaded at build time via scripts/download-wasm.sh)
+src/main/resources/wasm/

CLAUDE.md

@@ -48,7 +48,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 ## Architecture Overview
 
-This repository contains the official Schematic Java SDK, supporting Java 8+. Schematic is a feature flag and product analytics service.
+This repository contains the official Schematic Java SDK, supporting Java 11+. Schematic is a feature flag and product analytics service.
 
 ### Key Components
 

README.md

@@ -1,6 +1,6 @@
 # Schematic Java Library
 
-The official Schematic Java library, supporting Java 8+.
+The official Schematic Java library, supporting Java 11+.
 
 ## Installation and Setup
 
@@ -159,6 +159,135 @@ user.put("user_id", "your-user-id");
 boolean flagValue = schematic.checkFlag("some-flag-key", company, user);
 ```
 
+## DataStream
+
+DataStream enables local flag evaluation by maintaining a WebSocket connection to Schematic and caching flag rules, company, and user data locally. This reduces latency and network calls for flag checks.
+
+### Key Features
+- **Real-Time Updates**: Automatically updates cached data when changes occur on the backend.
+- **Configurable Caching**: Supports both in-memory (local) caching and Redis-based caching.
+- **Efficient Flag Checks**: Flag evaluation happens locally using a WASM rules engine.
+
+### Setup
+
+```java
+import com.schematic.api.Schematic;
+import com.schematic.api.datastream.DatastreamOptions;
+
+Schematic schematic = Schematic.builder()
+    .apiKey("YOUR_API_KEY")
+    .datastreamOptions(DatastreamOptions.builder()
+        .build())
+    .build();
+
+// Flag checks are now evaluated locally
+boolean flagValue = schematic.checkFlag("some-flag-key", company, user);
+
+// When done, close the client to release resources
+schematic.close();
+```
+
+### Configuration Options
+
+| Option | Type | Default | Description |
+|---|---|---|---|
+| `cacheTTL` | `Duration` | 24 hours | Cache TTL for flag/company/user data |
+| `redisCache` | `RedisCacheConfig` | — | Redis connection config (uses in-memory cache if not provided) |
+
+### Configuring Redis Cache
+
+DataStream supports Redis for caching, which is required for [Replicator Mode](#replicator-mode). Pass a `RedisCacheConfig` and the SDK will create and manage the Redis connection internally:
+
+```java
+import com.schematic.api.Schematic;
+import com.schematic.api.cache.RedisCacheConfig;
+import com.schematic.api.datastream.DatastreamOptions;
+import java.time.Duration;
+
+Schematic schematic = Schematic.builder()
+    .apiKey("YOUR_API_KEY")
+    .datastreamOptions(DatastreamOptions.builder()
+        .redisCache(RedisCacheConfig.builder()
+            .endpoint("localhost:6379")
+            .keyPrefix("schematic:")
+            .build())
+        .cacheTTL(Duration.ofMinutes(5))
+        .build())
+    .build();
+```
+
+#### Redis Configuration Options
+
+| Option | Type | Default | Description |
+|---|---|---|---|
+| `endpoint` | `String` | `localhost:6379` | Redis server address in `host:port` format |
+| `endpoints` | `List<String>` | `["localhost:6379"]` | Multiple endpoints (for future cluster support) |
+| `username` | `String` | — | Redis 6.0+ ACL username |
+| `password` | `String` | — | Redis password |
+| `database` | `int` | `0` | Redis database index |
+| `ssl` | `boolean` | `false` | Enable SSL/TLS |
+| `keyPrefix` | `String` | `schematic:` | Prefix for all Redis cache keys |
+| `connectTimeout` | `Duration` | 5 seconds | Connection timeout |
+| `readTimeout` | `Duration` | 3 seconds | Read timeout |
+| `maxPoolSize` | `int` | 8 | Maximum connection pool size |
+
+## Replicator Mode
+
+Replicator mode is designed for environments where a separate process (the [schematic-datastream-replicator](https://github.com/SchematicHQ/schematic-datastream-replicator)) manages the WebSocket connection and populates a shared Redis cache. The SDK reads from that cache and evaluates flags locally without establishing its own WebSocket connection.
+
+### Requirements
+
+Replicator mode **requires Redis** as a shared cache so the SDK can read data written by the external replicator process. An in-memory cache will not work since the replicator and SDK run in separate processes.
+
+### Setup
+
+```java
+import com.schematic.api.Schematic;
+import com.schematic.api.cache.RedisCacheConfig;
+import com.schematic.api.datastream.DatastreamOptions;
+
+Schematic schematic = Schematic.builder()
+    .apiKey("YOUR_API_KEY")
+    .datastreamOptions(DatastreamOptions.builder()
+        .redisCache(RedisCacheConfig.builder()
+            .endpoint("localhost:6379")
+            .build())
+        .withReplicatorMode("http://localhost:8090/ready")
+        .build())
+    .build();
+```
+
+### Configuration Options
+
+| Option | Type | Default | Description |
+|---|---|---|---|
+| `withReplicatorMode` | `String` | — | Enables replicator mode with the given health check URL |
+| `redisCache` | `RedisCacheConfig` | — | **Required.** Redis connection config for the shared cache |
+| `replicatorHealthCheckInterval` | `Duration` | 30 seconds | Health check polling interval |
+| `cacheTTL` | `Duration` | 24 hours | Cache TTL (should match the replicator's TTL) |
+
+### Cache TTL Configuration
+
+**Important:** 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 (e.g. after track events) will be written back with the shorter TTL and eventually evicted from the shared Redis cache, even though the replicator originally set them with no expiration.
+
+To match the replicator's default unlimited TTL:
+
+```java
+DatastreamOptions.builder()
+    .redisCache(RedisCacheConfig.builder()
+        .endpoint("localhost:6379")
+        .build())
+    .withReplicatorMode("http://localhost:8090/ready")
+    .cacheTTL(Duration.ZERO) // Unlimited, matching the replicator default
+    .build()
+```
+
+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 Java SDK provides utility functions to verify these signatures.

WASM_VERSION

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

build.gradle

@@ -18,6 +18,9 @@ dependencies {
     api 'com.fasterxml.jackson.core:jackson-databind:2.18.6'
     api 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.18.6'
     api 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.18.6'
+    implementation 'com.dylibso.chicory:runtime:1.4.0'
+    implementation 'com.dylibso.chicory:wasi:1.4.0'
+    implementation 'redis.clients:jedis:5.2.0'
     testImplementation 'org.junit.jupiter:junit-jupiter-api:5.8.2'
     testImplementation 'org.junit.jupiter:junit-jupiter-engine:5.8.2'
     testImplementation 'org.junit.jupiter:junit-jupiter-params:5.8.2'
@@ -26,8 +29,8 @@ dependencies {
 }
 
 
-sourceCompatibility = 1.8
-targetCompatibility = 1.8
+sourceCompatibility = 11
+targetCompatibility = 11
 
 tasks.withType(Javadoc) {
     failOnError false