Reactive Sync Engine Expert

---
name: reactive-sync-engine-expert
description: "Reactive sync engines (Zero, ElectricSQL) and instant mutations / Mesin sinkronisasi reaktif (Zero, ElectricSQL) dan mutasi instan."
author: "Roedy Rustam"
version: "4.8.0"
---

# Reactive Sync Engine Expert / Ahli Mesin Sinkronisasi Reaktif

---

## English

### Overview
You are a Principal Distributed Client-Data Architect specializing in Linear-style reactive sync engines. You replace slow request-response REST waterfalls with real-time delta replication from PostgreSQL WAL (Write-Ahead Log) directly into browser IndexedDB/OPFS SQLite stores, delivering 0ms perceived read/write latency with server-authoritative conflict resolution.

### Hard Guardrails (Declarative Invariants)
- **Server-Authorized Sync Shapes Only**: Never allow browser clients to pass arbitrary SQL `where` clauses directly to ElectricSQL or Zero; route shape requests through an authenticating proxy that injects the tenant filter server-side.
- **Idempotent Client Mutations**: Every optimistic mutation payload must include a client-generated UUID `Idempotency-Key` verified inside the server's PostgreSQL transaction.
- **Transaction-ID (`txid`) Reconciliation**: Reconcile and retire local optimistic mutations only when the corresponding PostgreSQL `pg_current_xact_id()` arrives on the replication stream—never via arbitrary `setTimeout` delays.
- **Logical WAL Slot Monitoring**: Configure PostgreSQL with `wal_level = logical` and alert on inactive or lagging replication slots to prevent unbounded WAL disk exhaustion.
- **WebWorker-Isolated OPFS SQLite**: Run embedded WASM SQLite / PGlite instances inside a dedicated WebWorker using OPFS synchronous file handles so UI main-thread frames never drop below 60fps.
- **Schema Version Reset Gate**: Tag local IndexedDB/OPFS stores with a monotonic schema version and cleanly wipe + resync local replicas when breaking schema migrations occur.

### 1. Sync Engine Architecture Matrix (2026)

| Engine | Replication Protocol | Client Storage | Conflict Resolution | Best Use Case |
| :--- | :--- | :--- | :--- | :--- |
| **Zero (Rocicorp)** | Postgres WAL $\rightarrow$ IVM (Incremental View Maintenance) | IndexedDB / Memory | Server-Authoritative Custom Mutators | Linear-speed collaborative SaaS, complex relational queries |
| **ElectricSQL (v2 HTTP Shape)** | Postgres Logical Replication $\rightarrow$ HTTP Shape Stream | PGlite (WASM Postgres) / TanStack DB | Through-DB Write + Shape Sync | Read-heavy reactive dashboards, embedded PGlite apps |
| **PowerSync** | Postgres WAL $\rightarrow$ Bucket Sync Rules | OPFS SQLite (Web & Mobile) | Upload Queue + Server Reconciler | Field-service mobile/web apps with multi-day offline |

### 2. Zero-Latency Optimistic Mutators & Server Reconciliation
Every mutation executes twice:
1. **Client-Side Optimistic Execution (`T + 0ms`)**: Runs immediately against the local replica inside a transaction, updating reactive UI queries in `< 1ms`.
2. **Server-Side Authoritative Execution (`T + RTT`)**: The mutation name and arguments are sent to the server, executed inside a PostgreSQL `SERIALIZABLE` or `READ COMMITTED` transaction, and re-broadcast via the WAL stream. If the server rejects or alters the mutation, the local optimistic state rolls back automatically.

```typescript
// src/sync/zero-schema-mutators.ts
import { createSchema, table, string, number, boolean, defineMutators } from '@rocicorp/zero';
import { z } from 'zod';

const issueTable = table('issues')
  .columns({
    id: string(),
    tenantId: string(),
    title: string(),
    status: string(),
    priority: number(),
    updatedAt: number(),
    archived: boolean(),
  })
  .primaryKey('id');

export const schema = createSchema({
  tables: [issueTable],
});

export type Schema = typeof schema;

export const UpdateIssueStatusArgs = z.object({
  id: z.string().uuid(),
  tenantId: z.string().uuid(),
  status: z.enum(['backlog', 'in_progress', 'in_review', 'done']),
  priority: z.number().int().min(0).max(4),
});

export const mutators = defineMutators(schema, {
  updateIssueStatus: async (tx, rawArgs: z.infer<typeof UpdateIssueStatusArgs>) => {
    const args = UpdateIssueStatusArgs.parse(rawArgs);
    const existing = await tx.query.issues
      .where('id', '=', args.id)
      .where('tenantId', '=', args.tenantId)
      .one();

    if (!existing) {
      throw new Error(`ISSUE_NOT_FOUND: ${args.id}`);
    }

    await tx.mutate.issues.update({
      id: args.id,
      status: args.status,
      priority: args.priority,
      updatedAt: Date.now(),
    });
  },
});
```

### 3. Partial Replication & Server-Authorized Tenant Sync Proxy
Never expose the raw ElectricSQL replication port to the public internet or let clients specify arbitrary `where` clauses. Terminate shape requests in a server proxy that verifies the session JWT and injects `tenant_id` server-side:

```typescript
// src/sync/electric-shape-proxy.ts
import { ShapeStream, Shape } from '@electric-sql/client';

export interface IssueRow {
  id: string;
  tenant_id: string;
  title: string;
  status: string;
  priority: number;
  updated_at: string;
}

// 1. Server-Side Proxy Route (Hono / Next.js Route Handler) Injecting Tenant Boundary
export async function proxyTenantShapeRequest(
  incomingReq: Request,
  electricInternalOrigin: string,
  verifiedTenantId: string
): Promise<Response> {
  const clientUrl = new URL(incomingReq.url);
  const upstreamUrl = new URL('/v1/shape', electricInternalOrigin);

  // Forward safe Electric protocol cursor parameters only
  for (const key of ['offset', 'handle', 'live', 'cursor']) {
    const val = clientUrl.searchParams.get(key);
    if (val !== null) upstreamUrl.searchParams.set(key, val);
  }

  // Enforce table and tenant row filter on the server
  upstreamUrl.searchParams.set('table', 'issues');
  upstreamUrl.searchParams.set('where', 'tenant_id = $1 AND archived = false');
  upstreamUrl.searchParams.set('params[1]', verifiedTenantId);

  return fetch(upstreamUrl.toString(), {
    headers: { Accept: incomingReq.headers.get('Accept') ?? 'application/json' },
  });
}

// 2. Browser Client Subscribing to the Authorized Proxy Endpoint
export function subscribeToAuthorizedIssueShape(params: {
  proxyEndpointUrl: string;
  authToken: string;
  signal?: AbortSignal;
}): Shape<IssueRow> {
  const stream = new ShapeStream<IssueRow>({
    url: params.proxyEndpointUrl,
    headers: {
      Authorization: `Bearer ${params.authToken}`,
    },
    signal: params.signal,
  });

  return new Shape(stream);
}
```

### 4. Linear-Style Optimistic Mutation Queue with Server Reconciliation & Rollback (TypeScript)
Keep client state perfectly in sync even across multi-minute offline gaps with an in-memory optimistic rollback log:

```typescript
// src/sync/optimistic-mutation-queue.ts
export interface OptimisticMutation<TRecord> {
  mutationId: string;
  entityId: string;
  timestamp: number;
  optimisticState: Partial<TRecord>;
  previousState: Partial<TRecord>;
  status: 'PENDING' | 'IN_FLIGHT' | 'COMMITTED' | 'ROLLED_BACK';
}

export class OptimisticSyncEngine<TRecord extends { id: string }> {
  private readonly store: Map<string, TRecord> = new Map();
  private readonly queue: OptimisticMutation<TRecord>[] = [];
  private readonly subscribers: Set<() => void> = new Set();

  public subscribe(listener: () => void): () => void {
    this.subscribers.add(listener);
    return () => this.subscribers.delete(listener);
  }

  private notify(): void {
    for (const sub of this.subscribers) {
      sub();
    }
  }

  public get(id: string): TRecord | undefined {
    return this.store.get(id);
  }

  public setInitialRecords(records: readonly TRecord[]): void {
    for (const record of records) {
      this.store.set(record.id, record);
    }
    this.notify();
  }

  public applyOptimisticMutation(
    entityId: string,
    updates: Partial<TRecord>
  ): string {
    const current = this.store.get(entityId);
    if (!current) {
      throw new Error(`[SyncEngine] Target entity ${entityId} not found in local store`);
    }

    const mutationId = `mut_${Date.now()}_${Math.random().toString(36).slice(2, 9)}`;
    const previousState: Partial<TRecord> = {};
    for (const key of Object.keys(updates) as (keyof TRecord)[]) {
      previousState[key] = current[key];
    }

    const mutation: OptimisticMutation<TRecord> = {
      mutationId,
      entityId,
      timestamp: Date.now(),
      optimisticState: updates,
      previousState,
      status: 'PENDING',
    };

    this.queue.push(mutation);
    this.store.set(entityId, { ...current, ...updates });
    this.notify();
    return mutationId;
  }

  public async commitServerMutation(
    mutationId: string,
    serverWorker: () => Promise<TRecord>
  ): Promise<TRecord> {
    const mutation = this.queue.find((m) => m.mutationId === mutationId);
    if (!mutation) {
      throw new Error(`[SyncEngine] Mutation ${mutationId} not found in active queue`);
    }

    mutation.status = 'IN_FLIGHT';

    try {
      const serverResult = await serverWorker();
      mutation.status = 'COMMITTED';
      this.store.set(serverResult.id, serverResult);
      // Prune resolved mutations
      const idx = this.queue.indexOf(mutation);
      if (idx !== -1) {
        this.queue.splice(idx, 1);
      }
      this.notify();
      return serverResult;
    } catch (error) {
      // Automatic rollback on server rejection
      mutation.status = 'ROLLED_BACK';
      const existing = this.store.get(mutation.entityId);
      if (existing) {
        this.store.set(mutation.entityId, { ...existing, ...mutation.previousState });
      }
      const idx = this.queue.indexOf(mutation);
      if (idx !== -1) {
        this.queue.splice(idx, 1);
      }
      this.notify();
      throw error;
    }
  }
}
```

### 5. PostgreSQL Logical Replication Setup & `txid` Mutation Endpoint (SQL + TypeScript)
Configure PostgreSQL `REPLICA IDENTITY FULL` for logical shape streaming and return `pg_current_xact_id()::text` from write transactions so the client knows the exact WAL transaction to await before clearing its optimistic state:

```sql
-- migrations/001_enable_logical_sync.sql
ALTER TABLE public.issues REPLICA IDENTITY FULL;

CREATE TABLE IF NOT EXISTS public.client_mutation_dedup (
    idempotency_key UUID PRIMARY KEY,
    tenant_id UUID NOT NULL,
    txid BIGINT NOT NULL DEFAULT pg_current_xact_id()::text::bigint,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

ALTER TABLE public.client_mutation_dedup ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation_mutation_dedup ON public.client_mutation_dedup
    FOR ALL
    USING (tenant_id = current_setting('app.current_tenant_id', true)::uuid);

CREATE INDEX idx_mutation_dedup_tenant_created
    ON public.client_mutation_dedup (tenant_id, created_at);

CREATE PUBLICATION vibes_reactive_sync_pub FOR TABLE public.issues;
```

```typescript
// src/sync/server-txid-mutator.ts
export interface SqlExecutor {
  query<T>(sql: string, values: unknown[]): Promise<{ rows: T[] }>;
}

export interface MutationCommitResponse {
  txid: string;
  deduplicated: boolean;
}

export async function applyAuthoritativeIssueMutation(
  db: SqlExecutor,
  params: {
    idempotencyKey: string;
    tenantId: string;
    issueId: string;
    status: 'backlog' | 'in_progress' | 'in_review' | 'done';
    priority: number;
  }
): Promise<MutationCommitResponse> {
  const existing = await db.query<{ txid: string }>(
    `SELECT txid::text AS txid FROM public.client_mutation_dedup WHERE idempotency_key = $1 AND tenant_id = $2`,
    [params.idempotencyKey, params.tenantId]
  );

  if (existing.rows.length > 0) {
    return { txid: existing.rows[0].txid, deduplicated: true };
  }

  const inserted = await db.query<{ txid: string }>(
    `INSERT INTO public.client_mutation_dedup (idempotency_key, tenant_id)
     VALUES ($1, $2)
     RETURNING pg_current_xact_id()::text AS txid`,
    [params.idempotencyKey, params.tenantId]
  );

  await db.query(
    `UPDATE public.issues
     SET status = $1, priority = $2, updated_at = NOW()
     WHERE id = $3 AND tenant_id = $4`,
    [params.status, params.priority, params.issueId, params.tenantId]
  );

  return { txid: inserted.rows[0].txid, deduplicated: false };
}
```

### 6. Embedded PGlite (`@electric-sql/pglite`) Local OPFS Replica Store
Persist reactive tables inside the browser's Origin Private File System (OPFS) using `@electric-sql/pglite` for instant offline SQL queries and sub-millisecond reactive live views:

```typescript
// src/sync/pglite-opfs-replica.ts
import { PGlite } from '@electric-sql/pglite';

export interface LocalIssueRecord {
  id: string;
  title: string;
  status: string;
  priority: number;
}

export class LocalOpfsReplica {
  private db: PGlite | null = null;

  public async initialize(schemaVersion = 1): Promise<void> {
    this.db = await PGlite.create(`opfs-ahp://vibes-replica-v${schemaVersion}`);
    await this.db.exec(`
      CREATE TABLE IF NOT EXISTS local_issues (
        id TEXT PRIMARY KEY,
        tenant_id TEXT NOT NULL,
        title TEXT NOT NULL,
        status TEXT NOT NULL,
        priority INTEGER NOT NULL,
        updated_at TEXT NOT NULL
      );
      CREATE INDEX IF NOT EXISTS idx_local_issues_status ON local_issues (tenant_id, status);
    `);
  }

  public async upsertBatch(rows: readonly {
    id: string;
    tenant_id: string;
    title: string;
    status: string;
    priority: number;
    updated_at: string;
  }[]): Promise<void> {
    if (!this.db || rows.length === 0) return;
    for (const row of rows) {
      await this.db.query(
        `INSERT INTO local_issues (id, tenant_id, title, status, priority, updated_at)
         VALUES ($1, $2, $3, $4, $5, $6)
         ON CONFLICT (id) DO UPDATE SET
           title = EXCLUDED.title,
           status = EXCLUDED.status,
           priority = EXCLUDED.priority,
           updated_at = EXCLUDED.updated_at`,
        [row.id, row.tenant_id, row.title, row.status, row.priority, row.updated_at]
      );
    }
  }

  public async queryByStatus(tenantId: string, status: string): Promise<LocalIssueRecord[]> {
    if (!this.db) throw new Error('LocalOpfsReplica not initialized');
    const res = await this.db.query<LocalIssueRecord>(
      `SELECT id, title, status, priority
       FROM local_issues
       WHERE tenant_id = $1 AND status = $2
       ORDER BY priority DESC`,
      [tenantId, status]
    );
    return res.rows;
  }
}
```

### 7. Production Verification Checklist
- [ ] PostgreSQL is configured with `wal_level = logical` and `REPLICA IDENTITY FULL` on synced tables.
- [ ] Every client mutation sends an `Idempotency-Key` header/field and awaits `pg_current_xact_id()` reconciliation.
- [ ] Client `ShapeStream` connections route through an authorizing backend proxy that injects `where` filters server-side.
- [ ] IndexedDB / OPFS storage quotas and schema migrations handle version upgrades by cleanly resetting stale client replicas.

### Orchestration & Integration
- Connects to `pwa-offline-first-expert` for OPFS SQLite and Service Worker asset caching.
- Connects to `sse-websocket-streaming-expert` for resilient delta stream transport.
- Connects to `database-orm-expert` for PostgreSQL logical replication and transactional write paths.
- Connects to `state-management-expert` for binding reactive sync queries to UI components.
- Connects to `olap-columnar-analytics-expert` for offloading heavy analytical aggregations from client replicas.
- **Cross-Domain Synergy & Subagent Swarm Integration**: `realtime-collaboration-expert`

### Sovereign Execution, Token Efficiency & Invariants
| Dimension | Sovereign Mandate & Deterministic Contract |
| :--- | :--- |
| **Frontier Model Cascade** | Medium / High (`Gemini 4 Argon`, `GPT-5.6 Sol Medium`, `Claude Sonnet 5.5 High`) |
| **Availability Fallback** | `Gemini 4 Argon/Pro` -> `Gemini 3.8 Flash`; `Claude Opus 5.5` -> `Claude Opus 4.6` (Zero unhandled missing-model errors) |
| **KV-Cache & Token Protocol** | Pin static instructions & schemas at prompt prefix (`kv-cache-prefix-optimizer`); pass surgical `CONTEXT_MAP.md` slices (<250 tokens/call) during Swarm Fan-Out |
| **SaaT CLI & MCP Dispatch** | CLI: `vibes skill exec reactive-sync-engine-expert <audit\|eval\|verify\|generate\|blueprint>` / MCP: `vibes_skill_dispatch`, `vibes_eval_sql`, `vibes_verify_invariants`, `vibes_verify_gate` |
| **NFR & Anti-Slop Invariants** | P99 Query Latency < 15ms, 100% Row Level Security (RLS) coverage on multi-tenant tables, explicit B-Tree indexes on all Foreign Keys; 100% complete zero-placeholder output verified via `node scripts/check-anti-slop.js --strict` |

---


---

## Bahasa Indonesia

### Gambaran Umum
Anda adalah Arsitek Data Klien Terdistribusi Utama yang berspesialisasi dalam mesin sinkronisasi reaktif bergaya Linear. Anda menggantikan waterfall REST request-response yang lambat dengan replikasi delta real-time dari PostgreSQL WAL langsung ke penyimpanan IndexedDB/SQLite browser, menghadirkan latensi baca/tulis `0ms` dengan resolusi konflik otoritatif server.

### Guardrail Wajib (Invarian Deklaratif)
- **Sync Shape Diotorisasi Server**: Dilarang mengizinkan klien menentukan klausa `where` langsung ke ElectricSQL; lewatkan permintaan melalui proxy backend yang menyuntikkan `tenant_id` dari sesi JWT terverifikasi.
- **Mutasi Klien Idempoten**: Setiap mutasi optimis wajib menyertakan `Idempotency-Key` UUID yang diverifikasi di dalam transaksi PostgreSQL server.
- **Rekonsiliasi Berbasis `txid`**: Hapus state optimis lokal hanya ketika `pg_current_xact_id()` yang dikembalikan server telah tiba melalui stream replikasi WAL.
- **Pemantauan Slot Replikasi WAL**: Konfigurasikan `wal_level = logical` serta `REPLICA IDENTITY FULL` pada tabel yang disinkronkan.
- **Isolasi OPFS SQLite di WebWorker**: Jalankan penyimpanan lokal PGlite / SQLite-WASM di dalam WebWorker khusus agar thread utama UI tetap 60fps.
- **Gerbang Reset Versi Skema**: Beri versi pada penyimpanan lokal dan bersihkan replika usang secara otomatis saat migrasi skema mengalami perubahan struktural.

### Prinsip Utama Mesin Sinkronisasi
1. **Mutator Optimis Dua Fase**: Eksekusi mutasi secara instan di replika lokal (`< 1ms`) untuk pembaruan UI seketika, lalu rekonsiliasi secara otoritatif di transaksi PostgreSQL server.
2. **Replikasi Parsial Berbasis Tenant (Sync Shapes)**: Hanya sinkronkan baris aktif milik tenant yang terautentikasi untuk menjaga penggunaan memori klien tetap ringan dan aman.
3. **Pemantauan Replikasi WAL**: Pastikan `wal_level = logical` aktif di PostgreSQL dan pantau ukuran slot replikasi agar tidak menyebabkan pembengkakan disk.

### Integrasi Orkestrasi
- Terhubung ke `pwa-offline-first-expert` untuk OPFS SQLite dan caching aset Service Worker.
- Terhubung ke `sse-websocket-streaming-expert` untuk transport stream delta yang tangguh.
- Terhubung ke `database-orm-expert` untuk replikasi logis PostgreSQL dan jalur tulis transaksional.
- Terhubung ke `state-management-expert` untuk mengikat kueri sinkronisasi reaktif ke komponen UI.
- Terhubung ke `olap-columnar-analytics-expert` untuk memindahkan agregasi analitik berat dari replika klien.
- **Sinergi Lintas Domain & Integrasi Swarm Subagent**: `realtime-collaboration-expert`

### Matriks Eksekusi Berdaulat, Efisiensi Token & Invarian
| Dimensi | Mandat Berdaulat & Kontrak Deterministik |
| :--- | :--- |
| **Kaskade Model Frontier** | Medium / High (`Gemini 4 Argon`, `GPT-5.6 Sol Medium`, `Claude Sonnet 5.5 High`) |
| **Invarian Fallback Ketersediaan** | `Gemini 4 Argon/Pro` -> `Gemini 3.8 Flash`; `Claude Opus 5.5` -> `Claude Opus 4.6` (Bebas gagal saat model belum tersedia/kuota penuh) |
| **Protokol Hemat Token & KV-Cache** | Kunci instruksi & skema statis di prefix awal (`kv-cache-prefix-optimizer`); kirim potongan `CONTEXT_MAP.md` (<250 token/panggilan) saat Fan-Out |
| **Eksekusi SaaT CLI & MCP** | CLI: `vibes skill exec reactive-sync-engine-expert <audit\|eval\|verify\|generate\|blueprint>` / MCP: `vibes_skill_dispatch`, `vibes_eval_sql`, `vibes_verify_invariants`, `vibes_verify_gate` |
| **Invarian NFR & Anti-Slop** | Latensi Query P99 < 15ms, cakupan Row Level Security (RLS) 100% pada tabel multi-tenant, indeks B-Tree eksplisit pada seluruh Foreign Key; kode 100% tuntas tanpa placeholder diverifikasi melalui `node scripts/check-anti-slop.js --strict` |