feat: implement vue-sync-engine with tab synchronization and transport layers
This commit is contained in:
@@ -0,0 +1,24 @@
|
||||
# Logs
|
||||
logs
|
||||
*.log
|
||||
npm-debug.log*
|
||||
yarn-debug.log*
|
||||
yarn-error.log*
|
||||
pnpm-debug.log*
|
||||
lerna-debug.log*
|
||||
|
||||
node_modules
|
||||
dist
|
||||
dist-ssr
|
||||
*.local
|
||||
|
||||
# Editor directories and files
|
||||
.vscode/*
|
||||
!.vscode/extensions.json
|
||||
.idea
|
||||
.DS_Store
|
||||
*.suo
|
||||
*.ntvs*
|
||||
*.njsproj
|
||||
*.sln
|
||||
*.sw?
|
||||
+3
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"recommendations": ["Vue.volar"]
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="UTF-8" />
|
||||
<link rel="icon" type="image/svg+xml" href="/favicon.svg" />
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
|
||||
<title>vue-sync-engine</title>
|
||||
</head>
|
||||
<body>
|
||||
<div id="app"></div>
|
||||
<script type="module" src="/src/main.ts"></script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -0,0 +1,27 @@
|
||||
{
|
||||
"name": "vue-sync-engine",
|
||||
"private": true,
|
||||
"version": "0.0.0",
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"dev": "vite",
|
||||
"build": "vue-tsc -b && vite build",
|
||||
"preview": "vite preview",
|
||||
"test": "vitest run",
|
||||
"test:watch": "vitest"
|
||||
},
|
||||
"dependencies": {
|
||||
"vue": "^3.5.34"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "^24.12.3",
|
||||
"@vitejs/plugin-vue": "^6.0.6",
|
||||
"@vue/test-utils": "^2.4.10",
|
||||
"@vue/tsconfig": "^0.9.1",
|
||||
"happy-dom": "^20.9.0",
|
||||
"typescript": "~6.0.2",
|
||||
"vite": "^8.0.12",
|
||||
"vitest": "^4.1.7",
|
||||
"vue-tsc": "^3.2.8"
|
||||
}
|
||||
}
|
||||
Generated
+1453
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,144 @@
|
||||
<script setup lang="ts">
|
||||
import { computed, ref } from 'vue'
|
||||
import { Status, useEngine, useInfiniteQuery, useMutation, useQuery } from './engine'
|
||||
import { PostEntity, UserEntity, postsInfinite, updatePostTitle, usersQuery, type Post, type User } from './demo.defs'
|
||||
import PostCard from './PostCard.vue'
|
||||
|
||||
const engine = useEngine()
|
||||
const selectedUserId = ref<number | undefined>(undefined)
|
||||
|
||||
const users = useQuery(usersQuery, () => undefined as void)
|
||||
const userList = computed(() =>
|
||||
(users.data.value?.ids ?? [])
|
||||
.map((id) => engine.mirror.getEntity<User>(UserEntity.name, id))
|
||||
.filter((u): u is User => !!u),
|
||||
)
|
||||
|
||||
const posts = useInfiniteQuery(postsInfinite, () => ({ userId: selectedUserId.value }))
|
||||
const postIds = computed(() => posts.pages.value.flatMap((p) => p.ids))
|
||||
const postsByIds = computed(() =>
|
||||
postIds.value
|
||||
.map((id) => engine.mirror.getEntity<Post>(PostEntity.name, id))
|
||||
.filter((p): p is Post => !!p),
|
||||
)
|
||||
|
||||
const editingId = ref<number | null>(null)
|
||||
const draft = ref('')
|
||||
const m = useMutation(updatePostTitle)
|
||||
|
||||
function startEdit(id: number, title: string) {
|
||||
editingId.value = id
|
||||
draft.value = title
|
||||
}
|
||||
|
||||
async function save() {
|
||||
if (editingId.value == null) return
|
||||
const id = editingId.value
|
||||
editingId.value = null
|
||||
try {
|
||||
await m.mutateAsync({ id, title: draft.value })
|
||||
} catch {}
|
||||
}
|
||||
</script>
|
||||
|
||||
<template>
|
||||
<main class="app">
|
||||
<header>
|
||||
<h1>vue-sync-engine demo</h1>
|
||||
<p>JSONPlaceholder · IndexedDB cache · optimistic mutations · infinite scroll</p>
|
||||
</header>
|
||||
|
||||
<aside class="users">
|
||||
<h2>Users <span v-if="users.isLoading.value">…</span></h2>
|
||||
<ul>
|
||||
<li>
|
||||
<button :class="{ active: selectedUserId === undefined }" @click="selectedUserId = undefined">
|
||||
All posts
|
||||
</button>
|
||||
</li>
|
||||
<li v-for="u in userList" :key="u.id">
|
||||
<button :class="{ active: selectedUserId === u.id }" @click="selectedUserId = u.id">
|
||||
{{ u.name }} <small>@{{ u.username }}</small>
|
||||
</button>
|
||||
</li>
|
||||
</ul>
|
||||
</aside>
|
||||
|
||||
<section class="posts">
|
||||
<h2>Posts <span v-if="posts.isLoading.value">…</span></h2>
|
||||
<PostCard
|
||||
v-for="p in postsByIds"
|
||||
:key="p.id"
|
||||
:post="p"
|
||||
:editing="editingId === p.id"
|
||||
:draft="draft"
|
||||
@edit="startEdit(p.id, p.title)"
|
||||
@input="draft = $event"
|
||||
@save="save"
|
||||
@cancel="editingId = null"
|
||||
/>
|
||||
<button class="more" :disabled="posts.isLoading.value" @click="posts.fetchNextPage">
|
||||
{{ posts.isLoading.value ? 'Loading…' : 'Load more' }}
|
||||
</button>
|
||||
<p v-if="m.status.value === Status.Error" class="err">Mutation failed: {{ m.error.value?.message }}</p>
|
||||
</section>
|
||||
</main>
|
||||
</template>
|
||||
|
||||
<style scoped>
|
||||
.app {
|
||||
max-width: 920px;
|
||||
margin: 0 auto;
|
||||
padding: 24px;
|
||||
font-family: system-ui, sans-serif;
|
||||
display: grid;
|
||||
grid-template-columns: 240px 1fr;
|
||||
gap: 24px;
|
||||
}
|
||||
header {
|
||||
grid-column: 1 / -1;
|
||||
}
|
||||
h1 { margin: 0; }
|
||||
h2 {
|
||||
margin-top: 0;
|
||||
font-size: 12px;
|
||||
text-transform: uppercase;
|
||||
letter-spacing: 0.08em;
|
||||
color: #666;
|
||||
}
|
||||
.users ul {
|
||||
list-style: none;
|
||||
padding: 0;
|
||||
margin: 0;
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 4px;
|
||||
}
|
||||
.users button {
|
||||
width: 100%;
|
||||
text-align: left;
|
||||
padding: 8px 10px;
|
||||
border: 1px solid #ddd;
|
||||
background: white;
|
||||
border-radius: 6px;
|
||||
cursor: pointer;
|
||||
font-size: 13px;
|
||||
}
|
||||
.users button.active {
|
||||
background: #111;
|
||||
color: white;
|
||||
border-color: #111;
|
||||
}
|
||||
.more {
|
||||
width: 100%;
|
||||
padding: 10px;
|
||||
border-radius: 6px;
|
||||
border: 1px dashed #ccc;
|
||||
background: #fafafa;
|
||||
cursor: pointer;
|
||||
}
|
||||
.err {
|
||||
color: #c00;
|
||||
font-size: 12px;
|
||||
}
|
||||
</style>
|
||||
@@ -0,0 +1,73 @@
|
||||
<script setup lang="ts">
|
||||
import type { Post } from './demo.defs'
|
||||
|
||||
defineProps<{ post: Post; editing: boolean; draft: string }>()
|
||||
defineEmits<{
|
||||
edit: []
|
||||
input: [value: string]
|
||||
save: []
|
||||
cancel: []
|
||||
}>()
|
||||
</script>
|
||||
|
||||
<template>
|
||||
<article class="post">
|
||||
<div v-if="editing" class="edit">
|
||||
<input :value="draft" @input="$emit('input', ($event.target as HTMLInputElement).value)" />
|
||||
<button @click="$emit('save')">Save</button>
|
||||
<button @click="$emit('cancel')">Cancel</button>
|
||||
</div>
|
||||
<h3 v-else>
|
||||
{{ post.title }}
|
||||
<button class="edit-btn" @click="$emit('edit')">✎</button>
|
||||
</h3>
|
||||
<p>{{ post.body }}</p>
|
||||
<small>user {{ post.userId }} · #{{ post.id }}</small>
|
||||
</article>
|
||||
</template>
|
||||
|
||||
<style scoped>
|
||||
.post {
|
||||
border: 1px solid #eee;
|
||||
border-radius: 8px;
|
||||
padding: 12px 16px;
|
||||
margin-bottom: 12px;
|
||||
}
|
||||
h3 {
|
||||
margin: 0 0 4px;
|
||||
font-size: 15px;
|
||||
display: flex;
|
||||
justify-content: space-between;
|
||||
align-items: center;
|
||||
gap: 8px;
|
||||
}
|
||||
p {
|
||||
margin: 0 0 6px;
|
||||
font-size: 13px;
|
||||
color: #444;
|
||||
}
|
||||
small {
|
||||
color: #999;
|
||||
font-size: 11px;
|
||||
}
|
||||
.edit {
|
||||
display: flex;
|
||||
gap: 6px;
|
||||
margin-bottom: 6px;
|
||||
}
|
||||
.edit input {
|
||||
flex: 1;
|
||||
padding: 6px 8px;
|
||||
font-size: 14px;
|
||||
}
|
||||
.edit-btn {
|
||||
background: transparent;
|
||||
border: none;
|
||||
cursor: pointer;
|
||||
opacity: 0.4;
|
||||
font-size: 14px;
|
||||
}
|
||||
.edit-btn:hover {
|
||||
opacity: 1;
|
||||
}
|
||||
</style>
|
||||
@@ -0,0 +1,87 @@
|
||||
import { defineEntity, defineInfiniteQuery, defineMutation, defineQuery, idbStore } from './engine'
|
||||
|
||||
export interface Post {
|
||||
id: number
|
||||
userId: number
|
||||
title: string
|
||||
body: string
|
||||
}
|
||||
|
||||
export interface User {
|
||||
id: number
|
||||
name: string
|
||||
email: string
|
||||
username: string
|
||||
}
|
||||
|
||||
export const PostEntity = defineEntity<Post>({
|
||||
name: 'post',
|
||||
id: (p) => p.id,
|
||||
storage: idbStore({ dbName: 'demo-sync-engine' }),
|
||||
})
|
||||
export const UserEntity = defineEntity<User>({
|
||||
name: 'user',
|
||||
id: (u) => u.id,
|
||||
storage: idbStore({ dbName: 'demo-sync-engine' }),
|
||||
})
|
||||
|
||||
const BASE = 'https://jsonplaceholder.typicode.com'
|
||||
|
||||
async function http<T>(url: string, init?: RequestInit, signal?: AbortSignal): Promise<T> {
|
||||
const res = await fetch(url, { ...init, signal })
|
||||
if (!res.ok) throw new Error(`HTTP ${res.status} for ${url}`)
|
||||
return (await res.json()) as T
|
||||
}
|
||||
|
||||
export const usersQuery = defineQuery<void, User[], { ids: number[] }>({
|
||||
name: 'users.list',
|
||||
key: () => ['users'],
|
||||
fetch: (_, ctx) => http<User[]>(`${BASE}/users`, undefined, ctx.signal),
|
||||
normalize: (items) => ({
|
||||
entities: { user: items },
|
||||
result: { ids: items.map((u) => u.id) },
|
||||
}),
|
||||
tags: () => ['users'],
|
||||
staleTime: 60_000,
|
||||
})
|
||||
|
||||
export const postsInfinite = defineInfiniteQuery<
|
||||
{ userId?: number },
|
||||
Post[],
|
||||
number,
|
||||
{ ids: number[]; nextPage: number | null }
|
||||
>({
|
||||
name: 'posts.infinite',
|
||||
key: (args) => ['posts', args.userId ?? 'all'],
|
||||
initialPageParam: 1,
|
||||
getNextPageParam: (last) => last.nextPage,
|
||||
fetch: (args, ctx) => {
|
||||
const params = new URLSearchParams({ _page: String(ctx.pageParam), _limit: '10' })
|
||||
if (args.userId != null) params.set('userId', String(args.userId))
|
||||
return http<Post[]>(`${BASE}/posts?${params}`, undefined, ctx.signal)
|
||||
},
|
||||
normalize: (items, _args, pageParam) => ({
|
||||
entities: { post: items },
|
||||
result: {
|
||||
ids: items.map((p) => p.id),
|
||||
nextPage: items.length === 10 ? (pageParam as number) + 1 : null,
|
||||
},
|
||||
}),
|
||||
tags: () => ['posts'],
|
||||
staleTime: 60_000,
|
||||
})
|
||||
|
||||
export const updatePostTitle = defineMutation<{ id: number; title: string }, Post>({
|
||||
name: 'post.updateTitle',
|
||||
fetch: (input, ctx) =>
|
||||
http<Post>(
|
||||
`${BASE}/posts/${input.id}`,
|
||||
{
|
||||
method: 'PATCH',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ title: input.title }),
|
||||
},
|
||||
ctx.signal,
|
||||
),
|
||||
optimistic: (input, ctx) => ctx.patchEntity(PostEntity, input.id, { title: input.title }),
|
||||
})
|
||||
@@ -0,0 +1,12 @@
|
||||
import { bootstrapWorker, indexedDBAdapter, createSharedWorkerServerEndpoint } from './engine'
|
||||
import registry from 'virtual:sync-engine-registry'
|
||||
|
||||
interface SharedWorkerScopeLike {
|
||||
onconnect: ((ev: { ports: readonly MessagePort[] }) => void) | null
|
||||
}
|
||||
|
||||
bootstrapWorker({
|
||||
...registry,
|
||||
storage: indexedDBAdapter({ dbName: 'demo-sync-engine' }),
|
||||
endpoint: createSharedWorkerServerEndpoint(self as unknown as SharedWorkerScopeLike),
|
||||
})
|
||||
@@ -0,0 +1,46 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { hashKey } from '../core/queryKey'
|
||||
import { applyPatch, invertEntityPatch } from '../core/patches'
|
||||
import { Op } from '../core/flags'
|
||||
|
||||
describe('queryKey.hashKey', () => {
|
||||
it('produces stable hash regardless of key order', () => {
|
||||
const a = hashKey(['users', { search: 'x', page: 1 }])
|
||||
const b = hashKey(['users', { page: 1, search: 'x' }])
|
||||
expect(a).toBe(b)
|
||||
})
|
||||
|
||||
it('different args produce different hashes', () => {
|
||||
expect(hashKey(['u', 1])).not.toBe(hashKey(['u', 2]))
|
||||
})
|
||||
})
|
||||
|
||||
describe('patches', () => {
|
||||
it('set at root', () => {
|
||||
expect(applyPatch({ a: 1 }, { op: Op.Set, path: [], value: { b: 2 } })).toEqual({ b: 2 })
|
||||
})
|
||||
|
||||
it('merge does not mutate input', () => {
|
||||
const input = { a: 1, b: 2 }
|
||||
const out = applyPatch(input, { op: Op.Merge, path: [], value: { b: 9 } })
|
||||
expect(out).toEqual({ a: 1, b: 9 })
|
||||
expect(input).toEqual({ a: 1, b: 2 })
|
||||
})
|
||||
|
||||
it('delete removes nested key', () => {
|
||||
const out = applyPatch({ a: { b: 1, c: 2 } }, { op: Op.Delete, path: ['a', 'b'] })
|
||||
expect(out).toEqual({ a: { c: 2 } })
|
||||
})
|
||||
|
||||
it('inverts a set on undefined prev as delete', () => {
|
||||
const inv = invertEntityPatch(undefined, { op: Op.Set, path: [], value: { x: 1 } })
|
||||
expect(inv).toEqual({ op: Op.Delete, path: [] })
|
||||
})
|
||||
|
||||
it('inverts a merge to previous slice', () => {
|
||||
const prev = { a: 1, b: 2 }
|
||||
const inv = invertEntityPatch(prev, { op: Op.Merge, path: [], value: { b: 9 } })
|
||||
expect(inv).toEqual({ op: Op.Merge, path: [], value: { b: 2 } })
|
||||
expect(applyPatch(applyPatch(prev, { op: Op.Merge, path: [], value: { b: 9 } }), inv)).toEqual(prev)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,219 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { effectScope } from 'vue'
|
||||
import { createInlineTransport } from '../transport/InlineTransport'
|
||||
import { createMirror } from '../tab/mirror'
|
||||
import { createTabRuntime } from '../tab/runtime'
|
||||
import { createQueryGraph, type AnyQueryDef } from '../worker/queryGraph'
|
||||
import { memoryAdapter } from '../adapters/storageAdapter'
|
||||
import { Status } from '../core/flags'
|
||||
import { flush, makeUserDefs, type ListUsersResp, type User, UserEntity } from './fixtures'
|
||||
|
||||
function setup(api: { list: any; update: any }) {
|
||||
const defs = makeUserDefs(api)
|
||||
const storage = memoryAdapter()
|
||||
const { client, server } = createInlineTransport()
|
||||
let onlineCb: (() => void) | null = null
|
||||
let online = true
|
||||
createQueryGraph({
|
||||
storage,
|
||||
endpoint: server,
|
||||
registry: {
|
||||
entities: new Map([[UserEntity.name, UserEntity]]),
|
||||
queries: new Map<string, AnyQueryDef>([
|
||||
[defs.usersList.name, defs.usersList],
|
||||
[defs.usersInfinite.name, defs.usersInfinite],
|
||||
]),
|
||||
mutations: new Map([[defs.updateUser.name, defs.updateUser]]),
|
||||
},
|
||||
isOnline: () => online,
|
||||
onOnline: (cb) => {
|
||||
onlineCb = cb
|
||||
return () => {}
|
||||
},
|
||||
})
|
||||
const mirror = createMirror()
|
||||
const runtime = createTabRuntime({ transport: client, mirror, staleSubGcMs: 10 })
|
||||
return {
|
||||
runtime,
|
||||
defs,
|
||||
storage,
|
||||
setOnline(v: boolean) {
|
||||
online = v
|
||||
if (v && onlineCb) onlineCb()
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
describe('useQuery + QueryGraph', () => {
|
||||
it('fetches, normalizes entities, and exposes result via mirror', async () => {
|
||||
const list = vi.fn(async (): Promise<ListUsersResp> => ({
|
||||
items: [
|
||||
{ id: '1', name: 'Ada', age: 30 },
|
||||
{ id: '2', name: 'Bob', age: 40 },
|
||||
],
|
||||
nextCursor: null,
|
||||
}))
|
||||
const { runtime, defs } = setup({ list, update: vi.fn() })
|
||||
|
||||
const scope = effectScope()
|
||||
let handle!: ReturnType<typeof runtime.subscribeQuery>
|
||||
scope.run(() => {
|
||||
handle = runtime.subscribeQuery(defs.usersList.name, defs.usersList.key({}), {})
|
||||
})
|
||||
|
||||
await flush()
|
||||
await flush()
|
||||
|
||||
const state = runtime.mirror.ensureQuery<{ ids: string[] }>(handle.subId)
|
||||
expect(state.value.status).toBe(Status.Success)
|
||||
expect(state.value.data).toEqual({ ids: ['1', '2'] })
|
||||
expect(runtime.mirror.getEntity<User>("user", "1")).toEqual({ id: '1', name: 'Ada', age: 30 })
|
||||
|
||||
scope.stop()
|
||||
})
|
||||
|
||||
it('dedupes parallel subscriptions to the same key (single fetch)', async () => {
|
||||
const list = vi.fn(async () => ({ items: [{ id: '1', name: 'A', age: 1 }], nextCursor: null }))
|
||||
const { runtime, defs } = setup({ list, update: vi.fn() })
|
||||
|
||||
const scope = effectScope()
|
||||
scope.run(() => {
|
||||
runtime.subscribeQuery(defs.usersList.name, defs.usersList.key({}), {})
|
||||
runtime.subscribeQuery(defs.usersList.name, defs.usersList.key({}), {})
|
||||
runtime.subscribeQuery(defs.usersList.name, defs.usersList.key({}), {})
|
||||
})
|
||||
|
||||
await flush()
|
||||
await flush()
|
||||
expect(list).toHaveBeenCalledTimes(1)
|
||||
scope.stop()
|
||||
})
|
||||
|
||||
it('hydrates from storage before network', async () => {
|
||||
const list = vi.fn(async () => ({ items: [{ id: '1', name: 'Fresh', age: 10 }], nextCursor: null }))
|
||||
const { runtime, defs, storage } = setup({ list, update: vi.fn() })
|
||||
|
||||
await storage.queries.write([{
|
||||
key: JSON.stringify(defs.usersList.key({})),
|
||||
value: {
|
||||
status: Status.Success,
|
||||
result: { ids: ['cached'] },
|
||||
updatedAt: Date.now() - 10_000,
|
||||
entityRefs: [],
|
||||
},
|
||||
}])
|
||||
|
||||
const scope = effectScope()
|
||||
let handle!: ReturnType<typeof runtime.subscribeQuery>
|
||||
scope.run(() => {
|
||||
handle = runtime.subscribeQuery(defs.usersList.name, defs.usersList.key({}), {})
|
||||
})
|
||||
|
||||
await flush()
|
||||
const state = runtime.mirror.ensureQuery<{ ids: string[] }>(handle.subId)
|
||||
expect(state.value.data).toEqual({ ids: ['cached'] })
|
||||
|
||||
await flush()
|
||||
await flush()
|
||||
expect(state.value.data).toEqual({ ids: ['1'] })
|
||||
scope.stop()
|
||||
})
|
||||
})
|
||||
|
||||
describe('useMutation + queue', () => {
|
||||
it('optimistic update is visible immediately, then confirmed by server response', async () => {
|
||||
const serverDb = new Map<string, User>([['1', { id: '1', name: 'A', age: 1 }]])
|
||||
const list = vi.fn(async () => ({ items: [...serverDb.values()], nextCursor: null }))
|
||||
const update = vi.fn(async (i: { id: string; patch: Partial<User> }) => {
|
||||
const next = { ...serverDb.get(i.id)!, ...i.patch }
|
||||
serverDb.set(i.id, next)
|
||||
return next
|
||||
})
|
||||
const { runtime, defs } = setup({ list, update })
|
||||
|
||||
const scope = effectScope()
|
||||
scope.run(() => runtime.subscribeQuery(defs.usersList.name, defs.usersList.key({}), {}))
|
||||
await flush()
|
||||
await flush()
|
||||
expect(runtime.mirror.getEntity<User>("user", "1")?.name).toBe('A')
|
||||
|
||||
const p = runtime.mutate(defs.updateUser.name, { id: '1', patch: { name: 'Renamed' } })
|
||||
await flush()
|
||||
expect(runtime.mirror.getEntity<User>("user", "1")?.name).toBe('Renamed')
|
||||
|
||||
await p
|
||||
await flush()
|
||||
await flush()
|
||||
expect(runtime.mirror.getEntity<User>("user", "1")?.name).toBe('Renamed')
|
||||
scope.stop()
|
||||
})
|
||||
|
||||
it('rolls back on server rejection', async () => {
|
||||
const list = vi.fn(async () => ({ items: [{ id: '1', name: 'A', age: 1 }], nextCursor: null }))
|
||||
const update = vi.fn(async () => {
|
||||
throw new Error('boom')
|
||||
})
|
||||
const { runtime, defs } = setup({ list, update })
|
||||
|
||||
const scope = effectScope()
|
||||
scope.run(() => runtime.subscribeQuery(defs.usersList.name, defs.usersList.key({}), {}))
|
||||
await flush()
|
||||
await flush()
|
||||
|
||||
await expect(
|
||||
runtime.mutate(defs.updateUser.name, { id: '1', patch: { name: 'Renamed' } }),
|
||||
).rejects.toThrow('boom')
|
||||
|
||||
expect(runtime.mirror.getEntity<User>("user", "1")?.name).toBe('A')
|
||||
scope.stop()
|
||||
})
|
||||
})
|
||||
|
||||
describe('useInfiniteQuery', () => {
|
||||
it('appends pages on fetchNextPage', async () => {
|
||||
let call = 0
|
||||
const list = vi.fn(async (args: { cursor?: string | null }): Promise<ListUsersResp> => {
|
||||
call++
|
||||
if (call === 1) return { items: [{ id: '1', name: 'A', age: 1 }], nextCursor: 'c1' }
|
||||
if (call === 2) return { items: [{ id: '2', name: 'B', age: 2 }], nextCursor: null }
|
||||
expect(args).toBeDefined()
|
||||
throw new Error('no more')
|
||||
})
|
||||
const { runtime, defs } = setup({ list, update: vi.fn() })
|
||||
|
||||
const scope = effectScope()
|
||||
let handle!: ReturnType<typeof runtime.subscribeQuery>
|
||||
scope.run(() => {
|
||||
handle = runtime.subscribeQuery(defs.usersInfinite.name, defs.usersInfinite.key({}), {})
|
||||
})
|
||||
await flush()
|
||||
await flush()
|
||||
|
||||
type R = { ids: string[]; nextCursor: string | null }
|
||||
const state = runtime.mirror.ensureQuery<{ pages: R[]; pageParams: unknown[] }>(handle.subId)
|
||||
expect(state.value.data?.pages).toEqual([{ ids: ['1'], nextCursor: 'c1' }])
|
||||
|
||||
handle.fetchNextPage()
|
||||
await flush()
|
||||
await flush()
|
||||
expect(state.value.data?.pages.length).toBe(2)
|
||||
expect(state.value.data?.pages[1].ids).toEqual(['2'])
|
||||
scope.stop()
|
||||
})
|
||||
})
|
||||
|
||||
describe('GC', () => {
|
||||
it('stops the scope after staleSubGcMs once refCount hits 0', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const list = vi.fn(async () => ({ items: [], nextCursor: null }))
|
||||
const { runtime, defs } = setup({ list, update: vi.fn() })
|
||||
const handle = runtime.subscribeQuery(defs.usersList.name, defs.usersList.key({}), {})
|
||||
handle.release()
|
||||
vi.advanceTimersByTime(20)
|
||||
expect(handle.scope.active).toBe(false)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,69 @@
|
||||
import { defineEntity, defineInfiniteQuery, defineMutation, defineQuery } from '../define'
|
||||
|
||||
export interface User {
|
||||
id: string
|
||||
name: string
|
||||
age: number
|
||||
}
|
||||
|
||||
export const UserEntity = defineEntity<User>({
|
||||
name: 'user',
|
||||
id: (u) => u.id,
|
||||
})
|
||||
|
||||
export interface ListUsersResp {
|
||||
items: User[]
|
||||
nextCursor: string | null
|
||||
}
|
||||
|
||||
export const flush = () =>
|
||||
new Promise<void>((r) =>
|
||||
queueMicrotask(() =>
|
||||
queueMicrotask(() =>
|
||||
queueMicrotask(() => queueMicrotask(() => queueMicrotask(r))),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
export function makeUserDefs(api: {
|
||||
list: (args: { search?: string; cursor?: string | null }) => Promise<ListUsersResp>
|
||||
update: (input: { id: string; patch: Partial<User> }) => Promise<User>
|
||||
}) {
|
||||
const usersList = defineQuery<{ search?: string }, ListUsersResp, { ids: string[] }>({
|
||||
name: 'users.list',
|
||||
key: (args) => ['users', 'list', args.search ?? ''],
|
||||
fetch: (args) => api.list({ search: args.search, cursor: null }),
|
||||
normalize: (resp) => ({
|
||||
entities: { user: resp.items },
|
||||
result: { ids: resp.items.map((u) => u.id) },
|
||||
}),
|
||||
tags: () => ['users'],
|
||||
staleTime: 1000,
|
||||
})
|
||||
|
||||
const usersInfinite = defineInfiniteQuery<
|
||||
{ search?: string },
|
||||
ListUsersResp,
|
||||
string | null,
|
||||
{ ids: string[]; nextCursor: string | null }
|
||||
>({
|
||||
name: 'users.infinite',
|
||||
key: (args) => ['users', 'infinite', args.search ?? ''],
|
||||
initialPageParam: null,
|
||||
getNextPageParam: (last) => last.nextCursor,
|
||||
fetch: (args, ctx) => api.list({ search: args.search, cursor: ctx.pageParam }),
|
||||
normalize: (resp) => ({
|
||||
entities: { user: resp.items },
|
||||
result: { ids: resp.items.map((u) => u.id), nextCursor: resp.nextCursor },
|
||||
}),
|
||||
})
|
||||
|
||||
const updateUser = defineMutation<{ id: string; patch: Partial<User> }, User>({
|
||||
name: 'users.update',
|
||||
fetch: (input) => api.update(input),
|
||||
optimistic: (input, ctx) => ctx.patchEntity(UserEntity, input.id, input.patch),
|
||||
invalidate: () => ['users'],
|
||||
})
|
||||
|
||||
return { usersList, usersInfinite, updateUser }
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
export interface StoreSpec {
|
||||
name: string
|
||||
keyPath?: string
|
||||
}
|
||||
|
||||
class IdbManager {
|
||||
readonly dbName: string
|
||||
private pending = new Map<string, StoreSpec>()
|
||||
private dbPromise: Promise<IDBDatabase> | null = null
|
||||
|
||||
constructor(dbName: string) {
|
||||
this.dbName = dbName
|
||||
}
|
||||
|
||||
registerStore(spec: StoreSpec | string): void {
|
||||
const s: StoreSpec = typeof spec === 'string' ? { name: spec } : spec
|
||||
const cur = this.pending.get(s.name)
|
||||
if (cur === undefined || (cur.keyPath === undefined && s.keyPath !== undefined)) {
|
||||
this.pending.set(s.name, s)
|
||||
}
|
||||
}
|
||||
|
||||
async getDb(): Promise<IDBDatabase> {
|
||||
if (this.dbPromise) {
|
||||
const db = await this.dbPromise
|
||||
const missing = this.missing(db)
|
||||
if (missing.length === 0) return db
|
||||
db.close()
|
||||
this.dbPromise = this.open(db.version + 1, missing)
|
||||
return this.dbPromise
|
||||
}
|
||||
this.dbPromise = (async () => {
|
||||
const initial = [...this.pending.values()]
|
||||
const db = await this.open(undefined, initial)
|
||||
const missing = this.missing(db)
|
||||
if (missing.length === 0) return db
|
||||
db.close()
|
||||
return this.open(db.version + 1, missing)
|
||||
})()
|
||||
return this.dbPromise
|
||||
}
|
||||
|
||||
async run<T>(
|
||||
storeName: string,
|
||||
mode: IDBTransactionMode,
|
||||
fn: (store: IDBObjectStore) => IDBRequest<T>,
|
||||
): Promise<T> {
|
||||
const db = await this.getDb()
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const tx = db.transaction(storeName, mode)
|
||||
const req = fn(tx.objectStore(storeName))
|
||||
req.onsuccess = () => resolve(req.result)
|
||||
req.onerror = () => reject(req.error)
|
||||
})
|
||||
}
|
||||
|
||||
async runTx(
|
||||
storeName: string,
|
||||
mode: IDBTransactionMode,
|
||||
fn: (store: IDBObjectStore) => void,
|
||||
): Promise<void> {
|
||||
const db = await this.getDb()
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
const tx = db.transaction(storeName, mode)
|
||||
fn(tx.objectStore(storeName))
|
||||
tx.oncomplete = () => resolve()
|
||||
tx.onerror = () => reject(tx.error)
|
||||
})
|
||||
}
|
||||
|
||||
private missing(db: IDBDatabase): StoreSpec[] {
|
||||
const out: StoreSpec[] = []
|
||||
for (const s of this.pending.values()) if (!db.objectStoreNames.contains(s.name)) out.push(s)
|
||||
return out
|
||||
}
|
||||
|
||||
private open(version: number | undefined, create: readonly StoreSpec[]): Promise<IDBDatabase> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const req = version === undefined ? indexedDB.open(this.dbName) : indexedDB.open(this.dbName, version)
|
||||
req.onupgradeneeded = () => {
|
||||
const db = req.result
|
||||
for (const s of create) {
|
||||
if (db.objectStoreNames.contains(s.name)) continue
|
||||
db.createObjectStore(s.name, s.keyPath ? { keyPath: s.keyPath } : undefined)
|
||||
}
|
||||
}
|
||||
req.onsuccess = () => {
|
||||
const db = req.result
|
||||
db.onversionchange = () => db.close()
|
||||
resolve(db)
|
||||
}
|
||||
req.onerror = () => reject(req.error)
|
||||
req.onblocked = () => reject(new Error(`IDB open blocked: ${this.dbName}`))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
const managers = new Map<string, IdbManager>()
|
||||
|
||||
export function getIdbManager(dbName: string): IdbManager {
|
||||
let m = managers.get(dbName)
|
||||
if (!m) {
|
||||
m = new IdbManager(dbName)
|
||||
managers.set(dbName, m)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
export type { IdbManager }
|
||||
@@ -0,0 +1,58 @@
|
||||
import type { EntityId } from '../core/types'
|
||||
import type { KeyedStore, KeyedStoreFactory } from '../core/keyedStore'
|
||||
import { getIdbManager } from './idbManager'
|
||||
|
||||
export interface IdbStoreOptions {
|
||||
dbName: string
|
||||
storeName?: string
|
||||
}
|
||||
|
||||
export function idbStore<T>(opts: IdbStoreOptions): KeyedStoreFactory<T> {
|
||||
const mgr = getIdbManager(opts.dbName)
|
||||
return (name) => {
|
||||
const store = opts.storeName ?? name
|
||||
mgr.registerStore(store)
|
||||
return {
|
||||
read(key: EntityId) {
|
||||
return mgr.run(store, 'readonly', (s) => s.get(asKey(key)) as IDBRequest<T | undefined>)
|
||||
},
|
||||
async readMany(keys: readonly EntityId[]) {
|
||||
if (keys.length === 0) return []
|
||||
const db = await mgr.getDb()
|
||||
return new Promise<Array<T | undefined>>((resolve, reject) => {
|
||||
const tx = db.transaction(store, 'readonly')
|
||||
const os = tx.objectStore(store)
|
||||
const out: Array<T | undefined> = new Array(keys.length)
|
||||
let pending = keys.length
|
||||
for (let i = 0; i < keys.length; i++) {
|
||||
const req = os.get(asKey(keys[i]))
|
||||
const idx = i
|
||||
req.onsuccess = () => {
|
||||
out[idx] = req.result as T | undefined
|
||||
if (--pending === 0) resolve(out)
|
||||
}
|
||||
req.onerror = () => reject(req.error)
|
||||
}
|
||||
})
|
||||
},
|
||||
readAll() {
|
||||
return mgr.run(store, 'readonly', (s) => s.getAll() as IDBRequest<T[]>)
|
||||
},
|
||||
write(items) {
|
||||
if (items.length === 0) return Promise.resolve()
|
||||
return mgr.runTx(store, 'readwrite', (os) => {
|
||||
for (let i = 0; i < items.length; i++) os.put(items[i].value, asKey(items[i].key))
|
||||
})
|
||||
},
|
||||
delete(key: EntityId) {
|
||||
return mgr.runTx(store, 'readwrite', (os) => {
|
||||
os.delete(asKey(key))
|
||||
})
|
||||
},
|
||||
} satisfies KeyedStore<T>
|
||||
}
|
||||
}
|
||||
|
||||
function asKey(k: EntityId): IDBValidKey {
|
||||
return typeof k === 'number' ? k : String(k)
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
import type { EntityId } from '../core/types'
|
||||
import type { KeyedStore, KeyedStoreFactory } from '../core/keyedStore'
|
||||
|
||||
export function memoryStore<T>(): KeyedStoreFactory<T> {
|
||||
return () => {
|
||||
const m = new Map<EntityId, T>()
|
||||
return {
|
||||
async read(key) {
|
||||
return m.get(key)
|
||||
},
|
||||
async readMany(keys) {
|
||||
const out: Array<T | undefined> = new Array(keys.length)
|
||||
for (let i = 0; i < keys.length; i++) out[i] = m.get(keys[i])
|
||||
return out
|
||||
},
|
||||
async readAll() {
|
||||
return [...m.values()]
|
||||
},
|
||||
async write(items) {
|
||||
for (let i = 0; i < items.length; i++) m.set(items[i].key, items[i].value)
|
||||
},
|
||||
async delete(key) {
|
||||
m.delete(key)
|
||||
},
|
||||
} satisfies KeyedStore<T>
|
||||
}
|
||||
}
|
||||
|
||||
export function noopStore<T>(): KeyedStoreFactory<T> {
|
||||
return () => noop as KeyedStore<T>
|
||||
}
|
||||
|
||||
const noop: KeyedStore<unknown> = {
|
||||
async read() {
|
||||
return undefined
|
||||
},
|
||||
async readMany(keys) {
|
||||
return new Array(keys.length).fill(undefined)
|
||||
},
|
||||
async readAll() {
|
||||
return []
|
||||
},
|
||||
async write() {},
|
||||
async delete() {},
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
import type { QueuedMutation, QuerySnapshot } from '../core/types'
|
||||
import type { KeyedStore } from '../core/keyedStore'
|
||||
import { memoryStore } from './memoryStore'
|
||||
import { idbStore } from './idbStore'
|
||||
|
||||
export interface StorageAdapter {
|
||||
queries: KeyedStore<QuerySnapshot>
|
||||
mutations: KeyedStore<QueuedMutation>
|
||||
}
|
||||
|
||||
export function memoryAdapter(): StorageAdapter {
|
||||
return {
|
||||
queries: memoryStore<QuerySnapshot>()('queries'),
|
||||
mutations: memoryStore<QueuedMutation>()('mutations'),
|
||||
}
|
||||
}
|
||||
|
||||
export interface IndexedDBAdapterOptions {
|
||||
dbName?: string
|
||||
}
|
||||
|
||||
export function indexedDBAdapter(opts: IndexedDBAdapterOptions = {}): StorageAdapter {
|
||||
const dbName = opts.dbName ?? 'sync-engine'
|
||||
return {
|
||||
queries: idbStore<QuerySnapshot>({ dbName })('queries'),
|
||||
mutations: idbStore<QueuedMutation>({ dbName })('mutations'),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
import { inject, type InjectionKey } from 'vue'
|
||||
import type { TabRuntime } from '../tab/runtime'
|
||||
|
||||
export const EngineKey: InjectionKey<TabRuntime> = Symbol('SyncEngine')
|
||||
|
||||
export function useEngine(): TabRuntime {
|
||||
const rt = inject(EngineKey)
|
||||
if (!rt) throw new Error('SyncEngine is not provided. Call app.provide(EngineKey, runtime).')
|
||||
return rt
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
import { computed, type ComputedRef, type MaybeRefOrGetter, toValue } from 'vue'
|
||||
import type { EntityDef, EntityId } from '../core/types'
|
||||
import { useEngine } from './useEngine'
|
||||
|
||||
export function useEntity<T>(
|
||||
def: EntityDef<T>,
|
||||
id: MaybeRefOrGetter<EntityId | undefined>,
|
||||
): ComputedRef<T | undefined> {
|
||||
const engine = useEngine()
|
||||
return computed(() => {
|
||||
const v = toValue(id)
|
||||
if (v === undefined || v === null) return undefined
|
||||
return engine.mirror.getEntity<T>(def.name, v)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
import { computed, onScopeDispose, watch, type ComputedRef, type MaybeRefOrGetter, toValue } from 'vue'
|
||||
import type { InfiniteQueryDef, QueryStatus } from '../core/types'
|
||||
import { Status } from '../core/flags'
|
||||
import { hashKey } from '../core/queryKey'
|
||||
import { useEngine } from './useEngine'
|
||||
|
||||
export interface UseInfiniteQueryReturn<TResult> {
|
||||
pages: ComputedRef<TResult[]>
|
||||
pageParams: ComputedRef<unknown[]>
|
||||
status: ComputedRef<QueryStatus>
|
||||
error: ComputedRef<{ message: string } | undefined>
|
||||
isLoading: ComputedRef<boolean>
|
||||
fetchNextPage: () => void
|
||||
}
|
||||
|
||||
interface InfinitePayload<T> {
|
||||
pages: T[]
|
||||
pageParams: unknown[]
|
||||
}
|
||||
|
||||
export function useInfiniteQuery<TArgs, TResp, TPageParam, TResult>(
|
||||
def: InfiniteQueryDef<TArgs, TResp, TPageParam, TResult> & { name: string },
|
||||
args: MaybeRefOrGetter<TArgs>,
|
||||
): UseInfiniteQueryReturn<TResult> {
|
||||
const engine = useEngine()
|
||||
|
||||
const initial = toValue(args)
|
||||
let handle = engine.subscribeQuery(def.name, def.key(initial), initial)
|
||||
let stateRef = engine.mirror.ensureQuery<InfinitePayload<TResult>>(handle.subId)
|
||||
|
||||
if (!def.staticHash) {
|
||||
watch(
|
||||
() => hashKey(def.key(toValue(args))),
|
||||
() => {
|
||||
const next = toValue(args)
|
||||
const prev = handle
|
||||
handle = engine.subscribeQuery(def.name, def.key(next), next)
|
||||
stateRef = engine.mirror.ensureQuery<InfinitePayload<TResult>>(handle.subId)
|
||||
prev.release()
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
onScopeDispose(() => handle.release())
|
||||
|
||||
return {
|
||||
pages: computed(() => stateRef.value.data?.pages ?? []),
|
||||
pageParams: computed(() => stateRef.value.data?.pageParams ?? []),
|
||||
status: computed(() => stateRef.value.status),
|
||||
error: computed(() => stateRef.value.error),
|
||||
isLoading: computed(() => stateRef.value.status === Status.Pending),
|
||||
fetchNextPage: () => handle.fetchNextPage(),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import { shallowRef, type ShallowRef } from 'vue'
|
||||
import type { MutationDef, QueryStatus } from '../core/types'
|
||||
import { Status } from '../core/flags'
|
||||
import { useEngine } from './useEngine'
|
||||
|
||||
export interface UseMutationReturn<TInput, TResp> {
|
||||
mutate: (input: TInput) => void
|
||||
mutateAsync: (input: TInput) => Promise<TResp>
|
||||
status: ShallowRef<QueryStatus>
|
||||
error: ShallowRef<Error | undefined>
|
||||
data: ShallowRef<TResp | undefined>
|
||||
}
|
||||
|
||||
export function useMutation<TInput, TResp>(
|
||||
def: MutationDef<TInput, TResp>,
|
||||
): UseMutationReturn<TInput, TResp> {
|
||||
const engine = useEngine()
|
||||
const status = shallowRef<QueryStatus>(Status.Idle)
|
||||
const error = shallowRef<Error | undefined>(undefined)
|
||||
const data = shallowRef<TResp | undefined>(undefined)
|
||||
|
||||
async function mutateAsync(input: TInput): Promise<TResp> {
|
||||
status.value = Status.Pending
|
||||
error.value = undefined
|
||||
try {
|
||||
const resp = (await engine.mutate(def.name, input)) as TResp
|
||||
data.value = resp
|
||||
status.value = Status.Success
|
||||
return resp
|
||||
} catch (e) {
|
||||
error.value = e as Error
|
||||
status.value = Status.Error
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
function mutate(input: TInput): void {
|
||||
void mutateAsync(input).catch(() => {})
|
||||
}
|
||||
|
||||
return { mutate, mutateAsync, status, error, data }
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
import { computed, onScopeDispose, watch, type ComputedRef, type MaybeRefOrGetter, toValue } from 'vue'
|
||||
import type { InfiniteQueryDef, QueryDef, QueryStatus } from '../core/types'
|
||||
import { Status } from '../core/flags'
|
||||
import { hashKey } from '../core/queryKey'
|
||||
import { useEngine } from './useEngine'
|
||||
|
||||
export interface UseQueryReturn<T> {
|
||||
data: ComputedRef<T | undefined>
|
||||
status: ComputedRef<QueryStatus>
|
||||
error: ComputedRef<{ message: string } | undefined>
|
||||
isLoading: ComputedRef<boolean>
|
||||
isSuccess: ComputedRef<boolean>
|
||||
isError: ComputedRef<boolean>
|
||||
}
|
||||
|
||||
export function useQuery<TArgs, TResp, TResult>(
|
||||
def: (QueryDef<TArgs, TResp, TResult> | InfiniteQueryDef<TArgs, TResp, any, TResult>) & { name: string },
|
||||
args: MaybeRefOrGetter<TArgs>,
|
||||
): UseQueryReturn<TResult> {
|
||||
const engine = useEngine()
|
||||
|
||||
const initial = toValue(args)
|
||||
let currentHandle = engine.subscribeQuery(def.name, def.key(initial), initial)
|
||||
let currentRef = engine.mirror.ensureQuery<TResult>(currentHandle.subId)
|
||||
|
||||
if (!def.staticHash) {
|
||||
watch(
|
||||
() => hashKey(def.key(toValue(args))),
|
||||
() => {
|
||||
const next = toValue(args)
|
||||
const prev = currentHandle
|
||||
currentHandle = engine.subscribeQuery(def.name, def.key(next), next)
|
||||
currentRef = engine.mirror.ensureQuery<TResult>(currentHandle.subId)
|
||||
prev.release()
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
onScopeDispose(() => currentHandle.release())
|
||||
|
||||
return {
|
||||
data: computed(() => currentRef.value.data),
|
||||
status: computed(() => currentRef.value.status),
|
||||
error: computed(() => currentRef.value.error),
|
||||
isLoading: computed(() => currentRef.value.status === Status.Pending),
|
||||
isSuccess: computed(() => currentRef.value.status === Status.Success),
|
||||
isError: computed(() => currentRef.value.status === Status.Error),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
export const Op = {
|
||||
Set: 1,
|
||||
Merge: 2,
|
||||
Delete: 4,
|
||||
} as const
|
||||
export type OpFlag = 1 | 2 | 4
|
||||
|
||||
export const Status = {
|
||||
Idle: 0,
|
||||
Pending: 1,
|
||||
Success: 2,
|
||||
Error: 3,
|
||||
} as const
|
||||
export type StatusFlag = 0 | 1 | 2 | 3
|
||||
|
||||
export const Msg = {
|
||||
Subscribe: 1,
|
||||
Unsubscribe: 2,
|
||||
Mutate: 3,
|
||||
FetchNextPage: 4,
|
||||
QueryPatch: 5,
|
||||
EntityPatch: 6,
|
||||
MutateResult: 7,
|
||||
} as const
|
||||
export type MsgKind = 1 | 2 | 3 | 4 | 5 | 6 | 7
|
||||
|
||||
export const Kind = {
|
||||
Entity: 1,
|
||||
Query: 2,
|
||||
Infinite: 3,
|
||||
Mutation: 4,
|
||||
} as const
|
||||
export type KindFlag = 1 | 2 | 3 | 4
|
||||
@@ -0,0 +1,11 @@
|
||||
import type { EntityId } from './types'
|
||||
|
||||
export interface KeyedStore<T = unknown> {
|
||||
read(key: EntityId): Promise<T | undefined>
|
||||
readMany(keys: readonly EntityId[]): Promise<Array<T | undefined>>
|
||||
readAll(): Promise<T[]>
|
||||
write(items: ReadonlyArray<{ key: EntityId; value: T }>): Promise<void>
|
||||
delete(key: EntityId): Promise<void>
|
||||
}
|
||||
|
||||
export type KeyedStoreFactory<T = unknown> = (name: string) => KeyedStore<T>
|
||||
@@ -0,0 +1,48 @@
|
||||
import type { Patch } from './types'
|
||||
import { Op } from './flags'
|
||||
|
||||
export function applyPatch<T>(target: T, patch: Patch): T {
|
||||
if (patch.path.length === 0) {
|
||||
if (patch.op === Op.Set) return patch.value as T
|
||||
if (patch.op === Op.Merge) return { ...(target as object), ...patch.value } as T
|
||||
return undefined as T
|
||||
}
|
||||
const next: any = Array.isArray(target) ? [...target] : { ...(target as any) }
|
||||
let cur = next
|
||||
for (let i = 0; i < patch.path.length - 1; i++) {
|
||||
const k = patch.path[i] as any
|
||||
const child = cur[k]
|
||||
cur[k] = Array.isArray(child) ? [...child] : { ...(child ?? {}) }
|
||||
cur = cur[k]
|
||||
}
|
||||
const last = patch.path[patch.path.length - 1] as any
|
||||
if (patch.op === Op.Set) cur[last] = patch.value
|
||||
else if (patch.op === Op.Merge) cur[last] = { ...(cur[last] ?? {}), ...patch.value }
|
||||
else cur[last] = undefined
|
||||
return next
|
||||
}
|
||||
|
||||
export function invertEntityPatch<T>(prev: T | undefined, patch: Patch): Patch {
|
||||
if (patch.op === Op.Set) {
|
||||
return prev === undefined
|
||||
? { op: Op.Delete, path: patch.path }
|
||||
: { op: Op.Set, path: patch.path, value: getAt(prev, patch.path) }
|
||||
}
|
||||
if (patch.op === Op.Delete) {
|
||||
return { op: Op.Set, path: patch.path, value: prev === undefined ? undefined : getAt(prev, patch.path) }
|
||||
}
|
||||
const prevSlice: Record<string, unknown> = {}
|
||||
for (const k of Object.keys(patch.value)) {
|
||||
prevSlice[k] = prev === undefined ? undefined : (getAt(prev, [...patch.path, k]) as unknown)
|
||||
}
|
||||
return { op: Op.Merge, path: patch.path, value: prevSlice }
|
||||
}
|
||||
|
||||
function getAt(obj: any, path: readonly (string | number)[]): unknown {
|
||||
let cur = obj
|
||||
for (const k of path) {
|
||||
if (cur == null) return undefined
|
||||
cur = cur[k as any]
|
||||
}
|
||||
return cur
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
import type { QueryKey } from './types'
|
||||
|
||||
export function hashKey(key: QueryKey): string {
|
||||
let s = '['
|
||||
for (let i = 0; i < key.length; i++) {
|
||||
if (i > 0) s += ','
|
||||
s += stringify(key[i])
|
||||
}
|
||||
return s + ']'
|
||||
}
|
||||
|
||||
export function entityKey(type: string, id: string | number): string {
|
||||
return `${type}\u0000${id}`
|
||||
}
|
||||
|
||||
function stringify(v: unknown): string {
|
||||
if (v === null) return 'null'
|
||||
const t = typeof v
|
||||
if (t === 'string') return JSON.stringify(v)
|
||||
if (t === 'number') return v === v && v !== Infinity && v !== -Infinity ? String(v) : 'null'
|
||||
if (t === 'boolean') return v ? 'true' : 'false'
|
||||
if (t === 'undefined') return 'null'
|
||||
if (Array.isArray(v)) {
|
||||
let s = '['
|
||||
for (let i = 0; i < v.length; i++) {
|
||||
if (i > 0) s += ','
|
||||
s += stringify(v[i])
|
||||
}
|
||||
return s + ']'
|
||||
}
|
||||
if (t === 'object') {
|
||||
const o = v as Record<string, unknown>
|
||||
const keys = Object.keys(o).sort()
|
||||
let s = '{'
|
||||
for (let i = 0; i < keys.length; i++) {
|
||||
if (i > 0) s += ','
|
||||
const k = keys[i]
|
||||
s += JSON.stringify(k) + ':' + stringify(o[k])
|
||||
}
|
||||
return s + '}'
|
||||
}
|
||||
return 'null'
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
export type EntityId = string | number
|
||||
|
||||
import type { StatusFlag, Kind } from './flags'
|
||||
import type { KeyedStore } from './keyedStore'
|
||||
|
||||
export interface EntityDef<T = any> {
|
||||
readonly kind: typeof Kind.Entity
|
||||
readonly name: string
|
||||
readonly id: (entity: T) => EntityId
|
||||
readonly storage?: KeyedStore<T>
|
||||
}
|
||||
|
||||
export interface NormalizedResult {
|
||||
entities: Record<string, ReadonlyArray<unknown>>
|
||||
result: unknown
|
||||
}
|
||||
|
||||
export interface ExecCtx {
|
||||
readonly signal: AbortSignal
|
||||
readonly pageParam: unknown
|
||||
}
|
||||
export interface ExecResult {
|
||||
readonly pageResult: unknown
|
||||
readonly entities: Record<string, ReadonlyArray<unknown>> | null
|
||||
}
|
||||
|
||||
export interface QueryDef<TArgs = any, TResp = any, TResult = any> {
|
||||
readonly kind: typeof Kind.Query
|
||||
readonly key: (args: TArgs) => readonly unknown[]
|
||||
readonly fetch: (args: TArgs, ctx: FetchCtx) => Promise<TResp>
|
||||
readonly normalize?: (resp: TResp, args: TArgs) => { entities?: Record<string, ReadonlyArray<unknown>>; result: TResult }
|
||||
readonly tags?: (args: TArgs) => readonly string[]
|
||||
readonly staleTime?: number
|
||||
readonly gcTime?: number
|
||||
readonly staticHash?: string | null
|
||||
readonly exec?: (args: TArgs, ctx: ExecCtx) => Promise<ExecResult>
|
||||
}
|
||||
|
||||
export interface InfiniteQueryDef<TArgs = any, TResp = any, TPageParam = any, TResult = any>
|
||||
extends Omit<QueryDef<TArgs, TResp, TResult>, 'kind' | 'fetch' | 'normalize' | 'exec'> {
|
||||
readonly kind: typeof Kind.Infinite
|
||||
readonly initialPageParam: TPageParam
|
||||
readonly getNextPageParam: (lastPage: TResult, allPages: TResult[]) => TPageParam | null | undefined
|
||||
readonly fetch: (args: TArgs, ctx: FetchCtx & { pageParam: TPageParam }) => Promise<TResp>
|
||||
readonly normalize?: (resp: TResp, args: TArgs, pageParam: TPageParam) => { entities?: Record<string, ReadonlyArray<unknown>>; result: TResult }
|
||||
readonly exec?: (args: TArgs, ctx: ExecCtx) => Promise<ExecResult>
|
||||
}
|
||||
|
||||
export interface MutationDef<TInput = any, TResp = any> {
|
||||
readonly kind: typeof Kind.Mutation
|
||||
readonly name: string
|
||||
readonly fetch: (input: TInput, ctx: FetchCtx) => Promise<TResp>
|
||||
readonly optimistic?: (input: TInput, ctx: OptimisticCtx) => void
|
||||
readonly onSuccess?: (resp: TResp, input: TInput, ctx: OptimisticCtx) => void
|
||||
readonly invalidate?: (input: TInput, resp?: TResp) => ReadonlyArray<QueryDef | InfiniteQueryDef | string>
|
||||
readonly maxRetries?: number
|
||||
}
|
||||
|
||||
export interface FetchCtx {
|
||||
readonly signal: AbortSignal
|
||||
}
|
||||
|
||||
export interface OptimisticCtx {
|
||||
patchEntity<T>(def: EntityDef<T>, id: EntityId, patch: Partial<T>): void
|
||||
removeEntity<T>(def: EntityDef<T>, id: EntityId): void
|
||||
upsertEntity<T>(def: EntityDef<T>, entity: T): void
|
||||
}
|
||||
|
||||
export type Patch =
|
||||
| { op: 1; path: readonly (string | number)[]; value: unknown }
|
||||
| { op: 2; path: readonly (string | number)[]; value: Record<string, unknown> }
|
||||
| { op: 4; path: readonly (string | number)[] }
|
||||
|
||||
export interface EntityPatch {
|
||||
type: string
|
||||
id: EntityId
|
||||
patch: Patch
|
||||
}
|
||||
|
||||
export type QueryStatus = StatusFlag
|
||||
|
||||
export interface QuerySnapshot<TResult = unknown> {
|
||||
status: QueryStatus
|
||||
result?: TResult
|
||||
error?: { message: string }
|
||||
updatedAt?: number
|
||||
entityRefs?: ReadonlyArray<{ type: string; id: EntityId }>
|
||||
}
|
||||
|
||||
export interface QueuedMutation {
|
||||
id: string
|
||||
seq: number
|
||||
name: string
|
||||
input: unknown
|
||||
inversePatches?: EntityPatch[]
|
||||
createdAt: number
|
||||
attempts: number
|
||||
state: 'pending' | 'inflight' | 'failed'
|
||||
}
|
||||
|
||||
export type QueryKey = readonly unknown[]
|
||||
@@ -0,0 +1,73 @@
|
||||
import type { App } from 'vue'
|
||||
import type { EntityDef, InfiniteQueryDef, MutationDef, QueryDef } from './core/types'
|
||||
import type { StorageAdapter } from './adapters/storageAdapter'
|
||||
import { memoryAdapter } from './adapters/storageAdapter'
|
||||
import { createInlineTransport } from './transport/InlineTransport'
|
||||
import { createQueryGraph } from './worker/queryGraph'
|
||||
import type { ServerEndpoint, Transport } from './transport/protocol'
|
||||
import { createMirror } from './tab/mirror'
|
||||
import { createTabRuntime, type TabRuntime } from './tab/runtime'
|
||||
import { EngineKey } from './composables/useEngine'
|
||||
|
||||
export interface WorkerBootstrapOptions {
|
||||
entities: ReadonlyArray<EntityDef>
|
||||
queries: ReadonlyArray<(QueryDef | InfiniteQueryDef) & { name: string }>
|
||||
mutations: ReadonlyArray<MutationDef>
|
||||
storage: StorageAdapter
|
||||
endpoint: ServerEndpoint
|
||||
defaultStaleTime?: number
|
||||
defaultGcTime?: number
|
||||
}
|
||||
|
||||
export function bootstrapWorker(opts: WorkerBootstrapOptions): void {
|
||||
const registry = {
|
||||
entities: new Map(opts.entities.map((e) => [e.name, e])),
|
||||
queries: new Map(opts.queries.map((q) => [q.name, q])),
|
||||
mutations: new Map(opts.mutations.map((m) => [m.name, m])),
|
||||
}
|
||||
createQueryGraph({
|
||||
storage: opts.storage,
|
||||
endpoint: opts.endpoint,
|
||||
registry,
|
||||
defaultStaleTime: opts.defaultStaleTime,
|
||||
defaultGcTime: opts.defaultGcTime,
|
||||
})
|
||||
}
|
||||
|
||||
export interface TabEngineOptions {
|
||||
transport: Transport
|
||||
staleSubGcMs?: number
|
||||
}
|
||||
|
||||
export function createTabEngine(opts: TabEngineOptions): TabRuntime {
|
||||
const mirror = createMirror()
|
||||
return createTabRuntime({ transport: opts.transport, mirror, staleSubGcMs: opts.staleSubGcMs })
|
||||
}
|
||||
|
||||
export interface EngineOptions {
|
||||
entities: ReadonlyArray<EntityDef>
|
||||
queries: ReadonlyArray<(QueryDef | InfiniteQueryDef) & { name: string }>
|
||||
mutations: ReadonlyArray<MutationDef>
|
||||
storage?: StorageAdapter
|
||||
defaultStaleTime?: number
|
||||
defaultGcTime?: number
|
||||
}
|
||||
|
||||
export function createEngine(opts: EngineOptions): TabRuntime {
|
||||
const storage = opts.storage ?? memoryAdapter()
|
||||
const { client, server } = createInlineTransport()
|
||||
bootstrapWorker({
|
||||
entities: opts.entities,
|
||||
queries: opts.queries,
|
||||
mutations: opts.mutations,
|
||||
storage,
|
||||
endpoint: server,
|
||||
defaultStaleTime: opts.defaultStaleTime,
|
||||
defaultGcTime: opts.defaultGcTime,
|
||||
})
|
||||
return createTabEngine({ transport: client })
|
||||
}
|
||||
|
||||
export function installEngine(app: App, runtime: TabRuntime): void {
|
||||
app.provide(EngineKey, runtime)
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
import type { EntityDef, ExecCtx, ExecResult, FetchCtx, InfiniteQueryDef, MutationDef, QueryDef } from './core/types'
|
||||
import type { KeyedStoreFactory } from './core/keyedStore'
|
||||
import { Kind } from './core/flags'
|
||||
import { hashKey } from './core/queryKey'
|
||||
|
||||
export function defineEntity<T>(def: {
|
||||
name: string
|
||||
id: (e: T) => string | number
|
||||
storage?: KeyedStoreFactory<T>
|
||||
}): EntityDef<T> {
|
||||
const storage = def.storage ? def.storage(def.name) : undefined
|
||||
return Object.freeze({ kind: Kind.Entity, name: def.name, id: def.id, storage })
|
||||
}
|
||||
|
||||
export function defineQuery<TArgs, TResp, TResult = TResp>(
|
||||
def: Omit<QueryDef<TArgs, TResp, TResult>, 'kind' | 'staticHash' | 'exec'> & { name: string },
|
||||
): QueryDef<TArgs, TResp, TResult> & { name: string } {
|
||||
return Object.freeze({
|
||||
kind: Kind.Query,
|
||||
...def,
|
||||
staticHash: precomputeStaticHash(def.key),
|
||||
exec: makeQueryExec<TArgs, TResp, TResult>(def.fetch, def.normalize),
|
||||
})
|
||||
}
|
||||
|
||||
export function defineInfiniteQuery<TArgs, TResp, TPageParam, TResult = TResp>(
|
||||
def: Omit<InfiniteQueryDef<TArgs, TResp, TPageParam, TResult>, 'kind' | 'staticHash' | 'exec'> & { name: string },
|
||||
): InfiniteQueryDef<TArgs, TResp, TPageParam, TResult> & { name: string } {
|
||||
return Object.freeze({
|
||||
kind: Kind.Infinite,
|
||||
...def,
|
||||
staticHash: precomputeStaticHash(def.key),
|
||||
exec: makeInfiniteExec<TArgs, TResp, TPageParam, TResult>(def.fetch, def.normalize),
|
||||
})
|
||||
}
|
||||
|
||||
export function defineMutation<TInput, TResp>(
|
||||
def: Omit<MutationDef<TInput, TResp>, 'kind'>,
|
||||
): MutationDef<TInput, TResp> {
|
||||
return Object.freeze({ kind: Kind.Mutation, ...def })
|
||||
}
|
||||
|
||||
function precomputeStaticHash(key: (args: any) => readonly unknown[]): string | null {
|
||||
if (key.length !== 0) return null
|
||||
try {
|
||||
return hashKey(key(undefined))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function makeQueryExec<TArgs, TResp, TResult>(
|
||||
fetch: (args: TArgs, ctx: FetchCtx) => Promise<TResp>,
|
||||
normalize?: (resp: TResp, args: TArgs) => { entities?: Record<string, ReadonlyArray<unknown>>; result: TResult },
|
||||
): (args: TArgs, ctx: ExecCtx) => Promise<ExecResult> {
|
||||
if (normalize) {
|
||||
return async (args, ctx) => {
|
||||
const resp = await fetch(args, { signal: ctx.signal })
|
||||
const norm = normalize(resp, args)
|
||||
return { pageResult: norm.result, entities: norm.entities ?? null }
|
||||
}
|
||||
}
|
||||
return async (args, ctx) => {
|
||||
const resp = await fetch(args, { signal: ctx.signal })
|
||||
return { pageResult: resp, entities: null }
|
||||
}
|
||||
}
|
||||
|
||||
function makeInfiniteExec<TArgs, TResp, TPageParam, TResult>(
|
||||
fetch: (args: TArgs, ctx: FetchCtx & { pageParam: TPageParam }) => Promise<TResp>,
|
||||
normalize?: (resp: TResp, args: TArgs, pageParam: TPageParam) => { entities?: Record<string, ReadonlyArray<unknown>>; result: TResult },
|
||||
): (args: TArgs, ctx: ExecCtx) => Promise<ExecResult> {
|
||||
if (normalize) {
|
||||
return async (args, ctx) => {
|
||||
const pp = ctx.pageParam as TPageParam
|
||||
const resp = await fetch(args, { signal: ctx.signal, pageParam: pp })
|
||||
const norm = normalize(resp, args, pp)
|
||||
return { pageResult: norm.result, entities: norm.entities ?? null }
|
||||
}
|
||||
}
|
||||
return async (args, ctx) => {
|
||||
const resp = await fetch(args, { signal: ctx.signal, pageParam: ctx.pageParam as TPageParam })
|
||||
return { pageResult: resp, entities: null }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
export * from './core/types'
|
||||
export type { KeyedStore, KeyedStoreFactory } from './core/keyedStore'
|
||||
export { hashKey, entityKey } from './core/queryKey'
|
||||
export { Op, Status, Msg, Kind } from './core/flags'
|
||||
export type { OpFlag, StatusFlag, MsgKind, KindFlag } from './core/flags'
|
||||
export { defineEntity, defineQuery, defineInfiniteQuery, defineMutation } from './define'
|
||||
export {
|
||||
createEngine,
|
||||
installEngine,
|
||||
bootstrapWorker,
|
||||
createTabEngine,
|
||||
type EngineOptions,
|
||||
type TabEngineOptions,
|
||||
type WorkerBootstrapOptions,
|
||||
} from './createEngine'
|
||||
export { EngineKey, useEngine } from './composables/useEngine'
|
||||
export { useQuery } from './composables/useQuery'
|
||||
export { useInfiniteQuery } from './composables/useInfiniteQuery'
|
||||
export { useEntity } from './composables/useEntity'
|
||||
export { useMutation } from './composables/useMutation'
|
||||
export type { StorageAdapter } from './adapters/storageAdapter'
|
||||
export { memoryAdapter, indexedDBAdapter, type IndexedDBAdapterOptions } from './adapters/storageAdapter'
|
||||
export { memoryStore, noopStore } from './adapters/memoryStore'
|
||||
export { idbStore, type IdbStoreOptions } from './adapters/idbStore'
|
||||
export { createInlineTransport } from './transport/InlineTransport'
|
||||
export { createSharedWorkerClientTransport, createSharedWorkerServerEndpoint } from './transport/SharedWorkerTransport'
|
||||
export type { Transport, ServerEndpoint, ClientMsg, ServerMsg } from './transport/protocol'
|
||||
export { createMirror } from './tab/mirror'
|
||||
export { createTabRuntime, type TabRuntime } from './tab/runtime'
|
||||
export { createQueryGraph } from './worker/queryGraph'
|
||||
export { syncEnginePlugin, type SyncEnginePluginOptions } from './plugin'
|
||||
@@ -0,0 +1,59 @@
|
||||
import type { Plugin } from 'vite'
|
||||
|
||||
const VIRTUAL_ID = 'virtual:sync-engine-registry'
|
||||
const RESOLVED_ID = '\0' + VIRTUAL_ID
|
||||
|
||||
export interface SyncEnginePluginOptions {
|
||||
definitions: string | readonly string[]
|
||||
}
|
||||
|
||||
export function syncEnginePlugin(opts: SyncEnginePluginOptions): Plugin {
|
||||
const patterns = Array.isArray(opts.definitions) ? opts.definitions : [opts.definitions]
|
||||
return {
|
||||
name: 'vue-sync-engine:registry',
|
||||
enforce: 'pre',
|
||||
resolveId(id) {
|
||||
if (id === VIRTUAL_ID) return RESOLVED_ID
|
||||
return null
|
||||
},
|
||||
load(id) {
|
||||
if (id !== RESOLVED_ID) return null
|
||||
const globs = patterns.map((p) => JSON.stringify(p)).join(', ')
|
||||
return `
|
||||
const KIND_ENTITY = 1
|
||||
const KIND_QUERY = 2
|
||||
const KIND_INFINITE = 3
|
||||
const KIND_MUTATION = 4
|
||||
const modules = import.meta.glob([${globs}], { eager: true })
|
||||
const entities = []
|
||||
const queries = []
|
||||
const mutations = []
|
||||
const seenEntities = new Set()
|
||||
const seenQueries = new Set()
|
||||
const seenMutations = new Set()
|
||||
for (const path in modules) {
|
||||
const mod = modules[path]
|
||||
for (const key in mod) {
|
||||
const v = mod[key]
|
||||
if (!v || typeof v !== 'object') continue
|
||||
const k = v.kind
|
||||
if (k === KIND_QUERY || k === KIND_INFINITE) {
|
||||
if (typeof v.name !== 'string' || seenQueries.has(v.name)) continue
|
||||
seenQueries.add(v.name)
|
||||
queries.push(v)
|
||||
} else if (k === KIND_MUTATION) {
|
||||
if (typeof v.name !== 'string' || seenMutations.has(v.name)) continue
|
||||
seenMutations.add(v.name)
|
||||
mutations.push(v)
|
||||
} else if (k === KIND_ENTITY) {
|
||||
if (typeof v.name !== 'string' || seenEntities.has(v.name)) continue
|
||||
seenEntities.add(v.name)
|
||||
entities.push(v)
|
||||
}
|
||||
}
|
||||
}
|
||||
export default { entities, queries, mutations }
|
||||
`
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
import { shallowRef, triggerRef, type ShallowRef } from 'vue'
|
||||
import type { EntityId, EntityPatch, Patch, QueryStatus } from '../core/types'
|
||||
import { Op, Status } from '../core/flags'
|
||||
import { applyPatch } from '../core/patches'
|
||||
|
||||
export interface QueryState<T = unknown> {
|
||||
status: QueryStatus
|
||||
data: T | undefined
|
||||
error: { message: string } | undefined
|
||||
}
|
||||
|
||||
export function createMirror() {
|
||||
const entities = new Map<string, Map<EntityId, unknown>>()
|
||||
const versions = new Map<string, ShallowRef<number>>()
|
||||
const queries = new Map<string, ShallowRef<QueryState>>()
|
||||
|
||||
function typeVersion(type: string): ShallowRef<number> {
|
||||
let v = versions.get(type)
|
||||
if (!v) {
|
||||
v = shallowRef(0)
|
||||
versions.set(type, v)
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
function entityBucket(type: string): Map<EntityId, unknown> {
|
||||
let b = entities.get(type)
|
||||
if (!b) {
|
||||
b = new Map()
|
||||
entities.set(type, b)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
function getEntity<T>(type: string, id: EntityId): T | undefined {
|
||||
typeVersion(type).value
|
||||
const b = entities.get(type)
|
||||
return b === undefined ? undefined : (b.get(id) as T | undefined)
|
||||
}
|
||||
|
||||
function applyEntityPatches(patches: EntityPatch[]): void {
|
||||
if (patches.length === 0) return
|
||||
let lastType = ''
|
||||
let bucket: Map<EntityId, unknown> | undefined
|
||||
let touchedFirst: string | undefined
|
||||
let touchedRest: Set<string> | undefined
|
||||
for (let i = 0; i < patches.length; i++) {
|
||||
const p = patches[i]
|
||||
if (p.type !== lastType) {
|
||||
lastType = p.type
|
||||
bucket = entityBucket(lastType)
|
||||
if (touchedFirst === undefined) touchedFirst = lastType
|
||||
else if (lastType !== touchedFirst) {
|
||||
if (touchedRest === undefined) touchedRest = new Set()
|
||||
touchedRest.add(lastType)
|
||||
}
|
||||
}
|
||||
const patch = p.patch
|
||||
if (patch.op === Op.Delete && patch.path.length === 0) {
|
||||
bucket!.delete(p.id)
|
||||
} else {
|
||||
bucket!.set(p.id, applyPatch(bucket!.get(p.id), patch))
|
||||
}
|
||||
}
|
||||
if (touchedFirst !== undefined) triggerRef(typeVersion(touchedFirst))
|
||||
if (touchedRest !== undefined) for (const t of touchedRest) triggerRef(typeVersion(t))
|
||||
}
|
||||
|
||||
function ensureQuery<T>(subId: string): ShallowRef<QueryState<T>> {
|
||||
let r = queries.get(subId) as ShallowRef<QueryState<T>> | undefined
|
||||
if (!r) {
|
||||
r = shallowRef<QueryState<T>>({ status: Status.Idle, data: undefined, error: undefined })
|
||||
queries.set(subId, r as ShallowRef<QueryState>)
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
function applyQueryPatch(subId: string, status: QueryStatus, patch?: Patch, error?: { message: string }): void {
|
||||
const r = ensureQuery(subId)
|
||||
const prev = r.value
|
||||
r.value = {
|
||||
status,
|
||||
data: patch ? applyPatch(prev.data, patch) : prev.data,
|
||||
error: error ?? prev.error,
|
||||
}
|
||||
}
|
||||
|
||||
function dropQuery(subId: string): void {
|
||||
queries.delete(subId)
|
||||
}
|
||||
|
||||
return { entities, getEntity, applyEntityPatches, ensureQuery, applyQueryPatch, dropQuery }
|
||||
}
|
||||
|
||||
export type Mirror = ReturnType<typeof createMirror>
|
||||
@@ -0,0 +1,113 @@
|
||||
import { effectScope, type EffectScope } from 'vue'
|
||||
import type { Transport } from '../transport/protocol'
|
||||
import type { Mirror } from './mirror'
|
||||
import { hashKey } from '../core/queryKey'
|
||||
import { Msg } from '../core/flags'
|
||||
|
||||
interface QuerySubHandle {
|
||||
subId: string
|
||||
refCount: number
|
||||
scope: EffectScope
|
||||
gcTimer: ReturnType<typeof setTimeout> | null
|
||||
release: () => void
|
||||
fetchNextPage: () => void
|
||||
}
|
||||
|
||||
export interface TabRuntime {
|
||||
mirror: Mirror
|
||||
transport: Transport
|
||||
subscribeQuery(defName: string, key: readonly unknown[], args: unknown): QuerySubHandle
|
||||
mutate(defName: string, input: unknown): Promise<unknown>
|
||||
dispose(): void
|
||||
}
|
||||
|
||||
export interface TabRuntimeOptions {
|
||||
transport: Transport
|
||||
mirror: Mirror
|
||||
staleSubGcMs?: number
|
||||
}
|
||||
|
||||
export function createTabRuntime(opts: TabRuntimeOptions): TabRuntime {
|
||||
const { transport, mirror } = opts
|
||||
const staleSubGcMs = opts.staleSubGcMs ?? 5_000
|
||||
|
||||
const byKey = new Map<string, QuerySubHandle>()
|
||||
const pendingMutations = new Map<string, { resolve: (v: unknown) => void; reject: (e: unknown) => void }>()
|
||||
const tabId =
|
||||
(typeof crypto !== 'undefined' && 'randomUUID' in crypto
|
||||
? crypto.randomUUID()
|
||||
: Math.random().toString(36).slice(2)) + '-'
|
||||
let subSeq = 0
|
||||
let mutSeq = 0
|
||||
|
||||
const off = transport.onMessage((msg) => {
|
||||
if (msg.type === Msg.QueryPatch) {
|
||||
mirror.applyQueryPatch(msg.subId, msg.status, msg.patch, msg.error)
|
||||
} else if (msg.type === Msg.EntityPatch) {
|
||||
mirror.applyEntityPatches(msg.patches)
|
||||
} else if (msg.type === Msg.MutateResult) {
|
||||
const p = pendingMutations.get(msg.mutId)
|
||||
if (p) {
|
||||
pendingMutations.delete(msg.mutId)
|
||||
if (msg.ok) p.resolve(msg.data)
|
||||
else p.reject(new Error(msg.error?.message ?? 'mutation failed'))
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
function subscribeQuery(defName: string, key: readonly unknown[], args: unknown): QuerySubHandle {
|
||||
const hash = hashKey(key)
|
||||
const existing = byKey.get(hash)
|
||||
if (existing) {
|
||||
if (existing.gcTimer !== null) {
|
||||
clearTimeout(existing.gcTimer)
|
||||
existing.gcTimer = null
|
||||
}
|
||||
existing.refCount++
|
||||
return existing
|
||||
}
|
||||
|
||||
const subId = `${tabId}s${++subSeq}`
|
||||
const scope = effectScope(true)
|
||||
mirror.ensureQuery(subId)
|
||||
transport.send({ type: Msg.Subscribe, subId, defName, args })
|
||||
|
||||
const handle: QuerySubHandle = {
|
||||
subId,
|
||||
refCount: 1,
|
||||
scope,
|
||||
gcTimer: null,
|
||||
fetchNextPage() {
|
||||
transport.send({ type: Msg.FetchNextPage, subId })
|
||||
},
|
||||
release() {
|
||||
handle.refCount--
|
||||
if (handle.refCount > 0) return
|
||||
handle.gcTimer = setTimeout(() => {
|
||||
byKey.delete(hash)
|
||||
transport.send({ type: Msg.Unsubscribe, subId })
|
||||
mirror.dropQuery(subId)
|
||||
scope.stop()
|
||||
}, staleSubGcMs)
|
||||
},
|
||||
}
|
||||
byKey.set(hash, handle)
|
||||
return handle
|
||||
}
|
||||
|
||||
function mutate(defName: string, input: unknown): Promise<unknown> {
|
||||
const mutId = `${tabId}m${++mutSeq}`
|
||||
return new Promise((resolve, reject) => {
|
||||
pendingMutations.set(mutId, { resolve, reject })
|
||||
transport.send({ type: Msg.Mutate, mutId, defName, input })
|
||||
})
|
||||
}
|
||||
|
||||
function dispose(): void {
|
||||
off()
|
||||
for (const h of byKey.values()) h.scope.stop()
|
||||
byKey.clear()
|
||||
}
|
||||
|
||||
return { mirror, transport, subscribeQuery, mutate, dispose }
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
import type { ClientMsg, ServerEndpoint, ServerMsg, Transport } from './protocol'
|
||||
|
||||
export function createInlineTransport(): { client: Transport; server: ServerEndpoint } {
|
||||
const clientHandlers = new Set<(m: ServerMsg) => void>()
|
||||
const serverHandlers = new Set<(m: ClientMsg) => void>()
|
||||
|
||||
let toServer: ClientMsg[] | null = null
|
||||
let toClient: ServerMsg[] | null = null
|
||||
|
||||
function drainToServer(): void {
|
||||
const batch = toServer!
|
||||
toServer = null
|
||||
for (let i = 0; i < batch.length; i++) for (const h of serverHandlers) h(batch[i])
|
||||
}
|
||||
|
||||
function drainToClient(): void {
|
||||
const batch = toClient!
|
||||
toClient = null
|
||||
for (let i = 0; i < batch.length; i++) for (const h of clientHandlers) h(batch[i])
|
||||
}
|
||||
|
||||
const client: Transport = {
|
||||
send(msg) {
|
||||
if (toServer) {
|
||||
toServer.push(msg)
|
||||
return
|
||||
}
|
||||
toServer = [msg]
|
||||
queueMicrotask(drainToServer)
|
||||
},
|
||||
onMessage(handler) {
|
||||
clientHandlers.add(handler)
|
||||
return () => clientHandlers.delete(handler)
|
||||
},
|
||||
}
|
||||
|
||||
const server: ServerEndpoint = {
|
||||
receive(msg) {
|
||||
for (const h of serverHandlers) h(msg)
|
||||
},
|
||||
broadcast(msg) {
|
||||
if (toClient) {
|
||||
toClient.push(msg)
|
||||
return
|
||||
}
|
||||
toClient = [msg]
|
||||
queueMicrotask(drainToClient)
|
||||
},
|
||||
onClient(handler) {
|
||||
serverHandlers.add(handler)
|
||||
return () => serverHandlers.delete(handler)
|
||||
},
|
||||
}
|
||||
|
||||
return { client, server }
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
import type { ClientMsg, ServerEndpoint, ServerMsg, Transport } from './protocol'
|
||||
|
||||
interface SharedWorkerLike {
|
||||
port: MessagePort
|
||||
}
|
||||
|
||||
interface SharedWorkerScopeLike {
|
||||
onconnect: ((ev: { ports: readonly MessagePort[] }) => void) | null
|
||||
}
|
||||
|
||||
export function createSharedWorkerClientTransport(worker: SharedWorkerLike): Transport {
|
||||
const handlers = new Set<(m: ServerMsg) => void>()
|
||||
worker.port.onmessage = (ev: MessageEvent<ServerMsg>) => {
|
||||
for (const h of handlers) h(ev.data)
|
||||
}
|
||||
worker.port.start()
|
||||
if (typeof window !== 'undefined') {
|
||||
window.addEventListener('beforeunload', () => {
|
||||
try {
|
||||
worker.port.close()
|
||||
} catch {}
|
||||
})
|
||||
}
|
||||
return {
|
||||
send(msg) {
|
||||
worker.port.postMessage(msg)
|
||||
},
|
||||
onMessage(handler) {
|
||||
handlers.add(handler)
|
||||
return () => handlers.delete(handler)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
export function createSharedWorkerServerEndpoint(scope: SharedWorkerScopeLike): ServerEndpoint {
|
||||
const ports = new Set<MessagePort>()
|
||||
const clientHandlers = new Set<(m: ClientMsg) => void>()
|
||||
|
||||
scope.onconnect = (ev) => {
|
||||
const port = ev.ports[0]
|
||||
ports.add(port)
|
||||
port.onmessage = (msg: MessageEvent<ClientMsg>) => {
|
||||
for (const h of clientHandlers) h(msg.data)
|
||||
}
|
||||
port.start()
|
||||
}
|
||||
|
||||
return {
|
||||
receive(msg) {
|
||||
for (const h of clientHandlers) h(msg)
|
||||
},
|
||||
broadcast(msg) {
|
||||
let dead: MessagePort[] | null = null
|
||||
for (const port of ports) {
|
||||
try {
|
||||
port.postMessage(msg)
|
||||
} catch {
|
||||
if (dead === null) dead = [port]
|
||||
else dead.push(port)
|
||||
}
|
||||
}
|
||||
if (dead !== null) for (let i = 0; i < dead.length; i++) ports.delete(dead[i])
|
||||
},
|
||||
onClient(handler) {
|
||||
clientHandlers.add(handler)
|
||||
return () => clientHandlers.delete(handler)
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
import type { EntityPatch, Patch, QueryStatus } from '../core/types'
|
||||
import { Msg } from '../core/flags'
|
||||
|
||||
export interface SubscribeMsg {
|
||||
type: typeof Msg.Subscribe
|
||||
subId: string
|
||||
defName: string
|
||||
args: unknown
|
||||
}
|
||||
export interface UnsubscribeMsg {
|
||||
type: typeof Msg.Unsubscribe
|
||||
subId: string
|
||||
}
|
||||
export interface MutateMsg {
|
||||
type: typeof Msg.Mutate
|
||||
mutId: string
|
||||
defName: string
|
||||
input: unknown
|
||||
}
|
||||
export interface FetchNextPageMsg {
|
||||
type: typeof Msg.FetchNextPage
|
||||
subId: string
|
||||
}
|
||||
|
||||
export type ClientMsg = SubscribeMsg | UnsubscribeMsg | MutateMsg | FetchNextPageMsg
|
||||
|
||||
export interface QueryPatchMsg {
|
||||
type: typeof Msg.QueryPatch
|
||||
subId: string
|
||||
status: QueryStatus
|
||||
patch?: Patch
|
||||
error?: { message: string }
|
||||
}
|
||||
export interface EntityPatchMsg {
|
||||
type: typeof Msg.EntityPatch
|
||||
patches: EntityPatch[]
|
||||
}
|
||||
export interface MutateResultMsg {
|
||||
type: typeof Msg.MutateResult
|
||||
mutId: string
|
||||
ok: boolean
|
||||
data?: unknown
|
||||
error?: { message: string }
|
||||
}
|
||||
|
||||
export type ServerMsg = QueryPatchMsg | EntityPatchMsg | MutateResultMsg
|
||||
|
||||
export interface Transport {
|
||||
send(msg: ClientMsg): void
|
||||
onMessage(handler: (msg: ServerMsg) => void): () => void
|
||||
}
|
||||
|
||||
export interface ServerEndpoint {
|
||||
receive(msg: ClientMsg): void
|
||||
broadcast(msg: ServerMsg): void
|
||||
onClient(handler: (msg: ClientMsg) => void): () => void
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
import type { StorageAdapter } from '../adapters/storageAdapter'
|
||||
import type { EntityPatch, MutationDef, OptimisticCtx, QueuedMutation } from '../core/types'
|
||||
|
||||
export interface MutationQueueDeps {
|
||||
storage: StorageAdapter
|
||||
mutations: Map<string, MutationDef>
|
||||
emitEntityPatches: (patches: EntityPatch[]) => void
|
||||
buildCtx: (forward: EntityPatch[], inverse: EntityPatch[]) => OptimisticCtx
|
||||
buildPostCtx: (post: EntityPatch[]) => OptimisticCtx
|
||||
invalidate: (def: MutationDef, input: unknown, resp: unknown) => void
|
||||
isOnline: () => boolean
|
||||
onOnline: (cb: () => void) => () => void
|
||||
onResult: (mutId: string, ok: boolean, data?: unknown, error?: { message: string }) => void
|
||||
}
|
||||
|
||||
interface InMemoryEntry {
|
||||
queued: QueuedMutation
|
||||
inverse: EntityPatch[]
|
||||
}
|
||||
|
||||
export function createMutationQueue(deps: MutationQueueDeps) {
|
||||
let seq = 0
|
||||
const inflight = new Map<string, InMemoryEntry>()
|
||||
let processing = false
|
||||
|
||||
function persist(m: QueuedMutation): Promise<void> {
|
||||
return deps.storage.mutations.write([{ key: m.id, value: m }])
|
||||
}
|
||||
|
||||
async function init(): Promise<void> {
|
||||
const persisted = await deps.storage.mutations.readAll()
|
||||
for (const m of persisted) {
|
||||
if (m.seq > seq) seq = m.seq
|
||||
inflight.set(m.id, { queued: m, inverse: m.inversePatches ?? [] })
|
||||
}
|
||||
void drain()
|
||||
deps.onOnline(() => void drain())
|
||||
}
|
||||
|
||||
async function enqueue(mutId: string, defName: string, input: unknown): Promise<void> {
|
||||
const def = deps.mutations.get(defName)
|
||||
if (!def) {
|
||||
if (__SYNC_ENGINE_DEV__) {
|
||||
deps.onResult(mutId, false, undefined, { message: `Unknown mutation: ${defName}` })
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
const forward: EntityPatch[] = []
|
||||
const inverse: EntityPatch[] = []
|
||||
if (def.optimistic) {
|
||||
def.optimistic(input, deps.buildCtx(forward, inverse))
|
||||
if (forward.length) deps.emitEntityPatches(forward)
|
||||
}
|
||||
|
||||
const queued: QueuedMutation = {
|
||||
id: mutId,
|
||||
seq: ++seq,
|
||||
name: defName,
|
||||
input,
|
||||
inversePatches: inverse,
|
||||
createdAt: Date.now(),
|
||||
attempts: 0,
|
||||
state: 'pending',
|
||||
}
|
||||
await persist(queued)
|
||||
inflight.set(mutId, { queued, inverse })
|
||||
void drain()
|
||||
}
|
||||
|
||||
async function drain(): Promise<void> {
|
||||
if (processing) return
|
||||
processing = true
|
||||
try {
|
||||
const ordered = [...inflight.values()].sort((a, b) => a.queued.seq - b.queued.seq)
|
||||
for (const entry of ordered) {
|
||||
if (!deps.isOnline()) break
|
||||
if (entry.queued.state === 'inflight') continue
|
||||
await runOne(entry)
|
||||
}
|
||||
} finally {
|
||||
processing = false
|
||||
}
|
||||
}
|
||||
|
||||
async function runOne(entry: InMemoryEntry): Promise<void> {
|
||||
const def = deps.mutations.get(entry.queued.name)
|
||||
if (!def) {
|
||||
inflight.delete(entry.queued.id)
|
||||
await deps.storage.mutations.delete(entry.queued.id)
|
||||
return
|
||||
}
|
||||
entry.queued.state = 'inflight'
|
||||
entry.queued.attempts++
|
||||
await persist(entry.queued)
|
||||
const ctrl = new AbortController()
|
||||
try {
|
||||
const resp = await def.fetch(entry.queued.input, { signal: ctrl.signal })
|
||||
if (def.onSuccess) {
|
||||
const post: EntityPatch[] = []
|
||||
def.onSuccess(resp, entry.queued.input, deps.buildPostCtx(post))
|
||||
if (post.length) deps.emitEntityPatches(post)
|
||||
}
|
||||
deps.invalidate(def, entry.queued.input, resp)
|
||||
inflight.delete(entry.queued.id)
|
||||
await deps.storage.mutations.delete(entry.queued.id)
|
||||
deps.onResult(entry.queued.id, true, resp)
|
||||
} catch (err) {
|
||||
const networkLike = !deps.isOnline() || isNetworkError(err)
|
||||
if (networkLike && entry.queued.attempts < (def.maxRetries ?? 5)) {
|
||||
entry.queued.state = 'pending'
|
||||
await persist(entry.queued)
|
||||
return
|
||||
}
|
||||
if (entry.inverse.length) deps.emitEntityPatches([...entry.inverse].reverse())
|
||||
inflight.delete(entry.queued.id)
|
||||
await deps.storage.mutations.delete(entry.queued.id)
|
||||
deps.onResult(entry.queued.id, false, undefined, { message: (err as Error)?.message ?? String(err) })
|
||||
}
|
||||
}
|
||||
|
||||
return { init, enqueue, drain }
|
||||
}
|
||||
|
||||
function isNetworkError(err: unknown): boolean {
|
||||
const msg = (err as Error)?.message?.toLowerCase() ?? ''
|
||||
return msg.includes('network') || msg.includes('fetch') || msg.includes('failed to fetch')
|
||||
}
|
||||
@@ -0,0 +1,449 @@
|
||||
import type { StorageAdapter } from '../adapters/storageAdapter'
|
||||
import type { EntityDef, EntityId, EntityPatch, InfiniteQueryDef, MutationDef, OptimisticCtx, QueryDef, QuerySnapshot, QueryStatus } from '../core/types'
|
||||
import { Op, Status, Msg, Kind } from '../core/flags'
|
||||
import { hashKey } from '../core/queryKey'
|
||||
import type { ServerEndpoint, ClientMsg } from '../transport/protocol'
|
||||
import { createMutationQueue } from './mutationQueue'
|
||||
|
||||
export type AnyQueryDef = (QueryDef | InfiniteQueryDef) & { name: string }
|
||||
|
||||
const EMPTY_PATH: readonly (string | number)[] = Object.freeze([])
|
||||
|
||||
interface QueryNode {
|
||||
key: string
|
||||
def: AnyQueryDef
|
||||
args: unknown
|
||||
subscribers: Set<string>
|
||||
status: QueryStatus
|
||||
result: unknown
|
||||
updatedAt: number
|
||||
inflight: Promise<void> | null
|
||||
abort: AbortController | null
|
||||
gcTimer: ReturnType<typeof setTimeout> | null
|
||||
entityRefs: Array<{ type: string; id: EntityId }>
|
||||
}
|
||||
|
||||
interface Registry {
|
||||
queries: Map<string, AnyQueryDef>
|
||||
mutations: Map<string, MutationDef>
|
||||
entities: Map<string, EntityDef>
|
||||
}
|
||||
|
||||
export interface QueryGraphOptions {
|
||||
storage: StorageAdapter
|
||||
endpoint: ServerEndpoint
|
||||
registry: Registry
|
||||
defaultStaleTime?: number
|
||||
defaultGcTime?: number
|
||||
isOnline?: () => boolean
|
||||
onOnline?: (cb: () => void) => () => void
|
||||
}
|
||||
|
||||
export function createQueryGraph(opts: QueryGraphOptions) {
|
||||
const { storage, endpoint, registry } = opts
|
||||
const defaultStaleTime = opts.defaultStaleTime ?? 30_000
|
||||
const defaultGcTime = opts.defaultGcTime ?? 5 * 60_000
|
||||
const isOnline = opts.isOnline ?? (() => (typeof navigator !== 'undefined' ? navigator.onLine : true))
|
||||
const onOnline =
|
||||
opts.onOnline ??
|
||||
((cb: () => void) => {
|
||||
if (typeof self === 'undefined') return () => {}
|
||||
self.addEventListener('online', cb)
|
||||
return () => self.removeEventListener('online', cb)
|
||||
})
|
||||
|
||||
const nodes = new Map<string, QueryNode>()
|
||||
const subToNode = new Map<string, QueryNode>()
|
||||
const entitiesInMemory = new Map<string, Map<EntityId, unknown>>()
|
||||
|
||||
function entityBucket(type: string): Map<EntityId, unknown> {
|
||||
let b = entitiesInMemory.get(type)
|
||||
if (!b) entitiesInMemory.set(type, (b = new Map()))
|
||||
return b
|
||||
}
|
||||
|
||||
function setEntity(type: string, id: EntityId, data: unknown): void {
|
||||
entityBucket(type).set(id, data)
|
||||
}
|
||||
|
||||
function getEntity(type: string, id: EntityId): unknown {
|
||||
return entityBucket(type).get(id)
|
||||
}
|
||||
|
||||
function emitEntityPatches(patches: EntityPatch[]): Promise<void> {
|
||||
if (patches.length === 0) return Promise.resolve()
|
||||
const writesByType = new Map<string, Array<{ key: EntityId; value: unknown }>>()
|
||||
const tasks: Promise<void>[] = []
|
||||
for (let i = 0; i < patches.length; i++) {
|
||||
const p = patches[i]
|
||||
const def = registry.entities.get(p.type)
|
||||
if (p.patch.op === Op.Delete) {
|
||||
if (def?.storage) tasks.push(def.storage.delete(p.id))
|
||||
} else if (def?.storage) {
|
||||
let arr = writesByType.get(p.type)
|
||||
if (!arr) {
|
||||
arr = []
|
||||
writesByType.set(p.type, arr)
|
||||
}
|
||||
arr.push({ key: p.id, value: getEntity(p.type, p.id) })
|
||||
}
|
||||
}
|
||||
for (const [type, writes] of writesByType) {
|
||||
const def = registry.entities.get(type)
|
||||
if (def?.storage) tasks.push(def.storage.write(writes))
|
||||
}
|
||||
endpoint.broadcast({ type: Msg.EntityPatch, patches })
|
||||
return tasks.length === 0 ? Promise.resolve() : Promise.all(tasks).then(noop)
|
||||
}
|
||||
|
||||
function mergeEntity(type: string, id: EntityId, data: unknown): EntityPatch | null {
|
||||
const prev = getEntity(type, id) as Record<string, unknown> | undefined
|
||||
if (prev && shallowEqual(prev, data as Record<string, unknown>)) return null
|
||||
setEntity(type, id, data)
|
||||
return { type, id, patch: { op: Op.Set, path: EMPTY_PATH, value: data } }
|
||||
}
|
||||
|
||||
function ingestEntities(
|
||||
buckets: Record<string, ReadonlyArray<unknown>>,
|
||||
refs?: Array<{ type: string; id: EntityId }>,
|
||||
): EntityPatch[] {
|
||||
const patches: EntityPatch[] = []
|
||||
for (const name in buckets) {
|
||||
const def = registry.entities.get(name)
|
||||
if (!def) continue
|
||||
const arr = buckets[name]
|
||||
for (let i = 0; i < arr.length; i++) {
|
||||
const e = arr[i]
|
||||
const id = def.id(e)
|
||||
if (refs) refs.push({ type: name, id })
|
||||
const p = mergeEntity(name, id, e)
|
||||
if (p) patches.push(p)
|
||||
}
|
||||
}
|
||||
return patches
|
||||
}
|
||||
|
||||
function ensureNode(defName: string, args: unknown): QueryNode {
|
||||
const def = registry.queries.get(defName)!
|
||||
if (__SYNC_ENGINE_DEV__ && !def) throw new Error(`Unknown query: ${defName}`)
|
||||
const key = def.staticHash ?? hashKey(def.key(args as never))
|
||||
let node = nodes.get(key)
|
||||
if (!node) {
|
||||
node = {
|
||||
key,
|
||||
def,
|
||||
args,
|
||||
subscribers: new Set(),
|
||||
status: Status.Idle,
|
||||
result: undefined,
|
||||
updatedAt: 0,
|
||||
inflight: null,
|
||||
abort: null,
|
||||
gcTimer: null,
|
||||
entityRefs: [],
|
||||
}
|
||||
nodes.set(key, node)
|
||||
} else if (node.gcTimer !== null) {
|
||||
clearTimeout(node.gcTimer)
|
||||
node.gcTimer = null
|
||||
}
|
||||
return node
|
||||
}
|
||||
|
||||
function scheduleGc(node: QueryNode): void {
|
||||
if (node.subscribers.size > 0) return
|
||||
const gc = node.def.gcTime ?? defaultGcTime
|
||||
node.gcTimer = setTimeout(() => {
|
||||
if (node.subscribers.size === 0) {
|
||||
nodes.delete(node.key)
|
||||
void storage.queries.delete(node.key)
|
||||
}
|
||||
}, gc)
|
||||
}
|
||||
|
||||
function isFresh(node: QueryNode): boolean {
|
||||
if (!node.updatedAt) return false
|
||||
const stale = node.def.staleTime ?? defaultStaleTime
|
||||
return Date.now() - node.updatedAt < stale
|
||||
}
|
||||
|
||||
async function hydrate(node: QueryNode): Promise<void> {
|
||||
const stored = await storage.queries.read(node.key)
|
||||
if (!stored || node.status !== Status.Idle) return
|
||||
if (!stored.entityRefs) {
|
||||
void storage.queries.delete(node.key)
|
||||
return
|
||||
}
|
||||
node.result = stored.result
|
||||
node.status = Status.Success
|
||||
node.updatedAt = stored.updatedAt
|
||||
if (stored.entityRefs.length > 0) {
|
||||
node.entityRefs = stored.entityRefs.slice()
|
||||
const patches = await loadEntityRefs(stored.entityRefs)
|
||||
if (patches.length > 0) endpoint.broadcast({ type: Msg.EntityPatch, patches })
|
||||
}
|
||||
pushSnapshotToSubscribers(node)
|
||||
}
|
||||
|
||||
async function loadEntityRefs(
|
||||
refs: ReadonlyArray<{ type: string; id: EntityId }>,
|
||||
): Promise<EntityPatch[]> {
|
||||
const byType = new Map<string, EntityId[]>()
|
||||
for (let i = 0; i < refs.length; i++) {
|
||||
const r = refs[i]
|
||||
let list = byType.get(r.type)
|
||||
if (!list) {
|
||||
list = []
|
||||
byType.set(r.type, list)
|
||||
}
|
||||
list.push(r.id)
|
||||
}
|
||||
const patches: EntityPatch[] = []
|
||||
for (const [type, ids] of byType) {
|
||||
const def = registry.entities.get(type)
|
||||
if (!def?.storage) continue
|
||||
const rows = await def.storage.readMany(ids)
|
||||
for (let i = 0; i < rows.length; i++) {
|
||||
const data = rows[i]
|
||||
if (data === undefined) continue
|
||||
const id = ids[i]
|
||||
if (getEntity(type, id) === undefined) setEntity(type, id, data)
|
||||
patches.push({ type, id, patch: { op: Op.Set, path: EMPTY_PATH, value: data } })
|
||||
}
|
||||
}
|
||||
return patches
|
||||
}
|
||||
|
||||
function pushSnapshotToSubscribers(node: QueryNode): void {
|
||||
for (const subId of node.subscribers) {
|
||||
endpoint.broadcast({
|
||||
type: Msg.QueryPatch,
|
||||
subId,
|
||||
status: node.status,
|
||||
patch: { op: Op.Set, path: EMPTY_PATH, value: node.result },
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
function broadcastEntityRefs(refs: ReadonlyArray<{ type: string; id: EntityId }>): void {
|
||||
if (refs.length === 0) return
|
||||
const patches: EntityPatch[] = []
|
||||
for (let i = 0; i < refs.length; i++) {
|
||||
const r = refs[i]
|
||||
const data = getEntity(r.type, r.id)
|
||||
if (data === undefined) continue
|
||||
patches.push({ type: r.type, id: r.id, patch: { op: Op.Set, path: EMPTY_PATH, value: data } })
|
||||
}
|
||||
if (patches.length > 0) endpoint.broadcast({ type: Msg.EntityPatch, patches })
|
||||
}
|
||||
|
||||
async function runFetch(node: QueryNode, pageParam?: unknown, append = false): Promise<void> {
|
||||
if (node.inflight) return node.inflight
|
||||
node.status = Status.Pending
|
||||
for (const subId of node.subscribers) {
|
||||
endpoint.broadcast({ type: Msg.QueryPatch, subId, status: Status.Pending })
|
||||
}
|
||||
node.abort = new AbortController()
|
||||
const isInfinite = node.def.kind === Kind.Infinite
|
||||
const effectivePageParam = isInfinite
|
||||
? pageParam ?? (node.def as InfiniteQueryDef).initialPageParam
|
||||
: undefined
|
||||
const exec = (async () => {
|
||||
try {
|
||||
const pageRefs: Array<{ type: string; id: EntityId }> = []
|
||||
const { pageResult, entities } = await node.def.exec!(node.args as never, {
|
||||
signal: node.abort!.signal,
|
||||
pageParam: effectivePageParam,
|
||||
})
|
||||
if (entities !== null) await emitEntityPatches(ingestEntities(entities, pageRefs))
|
||||
if (isInfinite) {
|
||||
const prev = (node.result as { pages: unknown[]; pageParams: unknown[] } | undefined) ?? { pages: [], pageParams: [] }
|
||||
node.result = append
|
||||
? { pages: [...prev.pages, pageResult], pageParams: [...prev.pageParams, effectivePageParam] }
|
||||
: { pages: [pageResult], pageParams: [effectivePageParam] }
|
||||
node.entityRefs = append ? node.entityRefs.concat(pageRefs) : pageRefs
|
||||
} else {
|
||||
node.result = pageResult
|
||||
node.entityRefs = pageRefs
|
||||
}
|
||||
node.status = Status.Success
|
||||
node.updatedAt = Date.now()
|
||||
const snap: QuerySnapshot = {
|
||||
status: Status.Success,
|
||||
result: node.result,
|
||||
updatedAt: node.updatedAt,
|
||||
entityRefs: node.entityRefs,
|
||||
}
|
||||
await storage.queries.write([{ key: node.key, value: snap }])
|
||||
pushSnapshotToSubscribers(node)
|
||||
} catch (err) {
|
||||
node.status = Status.Error
|
||||
const error = { message: (err as Error)?.message ?? String(err) }
|
||||
for (const subId of node.subscribers) {
|
||||
endpoint.broadcast({ type: Msg.QueryPatch, subId, status: Status.Error, error })
|
||||
}
|
||||
} finally {
|
||||
node.inflight = null
|
||||
node.abort = null
|
||||
}
|
||||
})()
|
||||
node.inflight = exec
|
||||
return exec
|
||||
}
|
||||
|
||||
function fetchNextPage(subId: string): void {
|
||||
const node = subToNode.get(subId)
|
||||
if (!node || node.def.kind !== Kind.Infinite) return
|
||||
const def = node.def as InfiniteQueryDef
|
||||
const cur = (node.result as { pages: unknown[]; pageParams: unknown[] } | undefined) ?? { pages: [], pageParams: [] }
|
||||
const last = cur.pages[cur.pages.length - 1]
|
||||
if (last === undefined) {
|
||||
void runFetch(node, def.initialPageParam, false)
|
||||
return
|
||||
}
|
||||
const next = def.getNextPageParam(last as never, cur.pages as never[])
|
||||
if (next === null || next === undefined) return
|
||||
void runFetch(node, next, true)
|
||||
}
|
||||
|
||||
async function subscribe(msg: { subId: string; defName: string; args: unknown }): Promise<void> {
|
||||
const node = ensureNode(msg.defName, msg.args)
|
||||
node.subscribers.add(msg.subId)
|
||||
subToNode.set(msg.subId, node)
|
||||
|
||||
if (node.status === Status.Success) {
|
||||
broadcastEntityRefs(node.entityRefs)
|
||||
endpoint.broadcast({
|
||||
type: Msg.QueryPatch,
|
||||
subId: msg.subId,
|
||||
status: Status.Success,
|
||||
patch: { op: Op.Set, path: EMPTY_PATH, value: node.result },
|
||||
})
|
||||
if (!isFresh(node)) void runFetch(node)
|
||||
return
|
||||
}
|
||||
if (node.status === Status.Idle) await hydrate(node)
|
||||
const status = node.status as QueryNode['status']
|
||||
if (status === Status.Pending) endpoint.broadcast({ type: Msg.QueryPatch, subId: msg.subId, status: Status.Pending })
|
||||
else if (status === Status.Success) {
|
||||
broadcastEntityRefs(node.entityRefs)
|
||||
endpoint.broadcast({
|
||||
type: Msg.QueryPatch,
|
||||
subId: msg.subId,
|
||||
status: Status.Success,
|
||||
patch: { op: Op.Set, path: EMPTY_PATH, value: node.result },
|
||||
})
|
||||
}
|
||||
if (!isFresh(node)) void runFetch(node)
|
||||
}
|
||||
|
||||
function unsubscribe(subId: string): void {
|
||||
const node = subToNode.get(subId)
|
||||
if (!node) return
|
||||
subToNode.delete(subId)
|
||||
node.subscribers.delete(subId)
|
||||
if (node.subscribers.size === 0) scheduleGc(node)
|
||||
}
|
||||
|
||||
function buildCtx(forward: EntityPatch[], inverse: EntityPatch[]): OptimisticCtx {
|
||||
return {
|
||||
patchEntity: (entDef, id, patch) => {
|
||||
const prev = getEntity(entDef.name, id) as Record<string, unknown> | undefined
|
||||
const next = { ...(prev ?? {}), ...(patch as Record<string, unknown>) }
|
||||
setEntity(entDef.name, id, next)
|
||||
forward.push({ type: entDef.name, id, patch: { op: Op.Merge, path: EMPTY_PATH, value: patch as Record<string, unknown> } })
|
||||
if (prev !== undefined) {
|
||||
const prevSlice: Record<string, unknown> = {}
|
||||
for (const k of Object.keys(patch as Record<string, unknown>)) prevSlice[k] = (prev as any)[k]
|
||||
inverse.push({ type: entDef.name, id, patch: { op: Op.Merge, path: EMPTY_PATH, value: prevSlice } })
|
||||
} else {
|
||||
inverse.push({ type: entDef.name, id, patch: { op: Op.Delete, path: EMPTY_PATH } })
|
||||
}
|
||||
},
|
||||
removeEntity: (entDef, id) => {
|
||||
const prev = getEntity(entDef.name, id)
|
||||
entityBucket(entDef.name).delete(id)
|
||||
forward.push({ type: entDef.name, id, patch: { op: Op.Delete, path: EMPTY_PATH } })
|
||||
if (prev !== undefined) inverse.push({ type: entDef.name, id, patch: { op: Op.Set, path: EMPTY_PATH, value: prev } })
|
||||
},
|
||||
upsertEntity: (entDef, entity) => {
|
||||
const id = entDef.id(entity)
|
||||
const prev = getEntity(entDef.name, id)
|
||||
setEntity(entDef.name, id, entity)
|
||||
forward.push({ type: entDef.name, id, patch: { op: Op.Set, path: EMPTY_PATH, value: entity } })
|
||||
if (prev === undefined) inverse.push({ type: entDef.name, id, patch: { op: Op.Delete, path: EMPTY_PATH } })
|
||||
else inverse.push({ type: entDef.name, id, patch: { op: Op.Set, path: EMPTY_PATH, value: prev } })
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function buildPostCtx(post: EntityPatch[]): OptimisticCtx {
|
||||
return {
|
||||
patchEntity: (entDef, id, patch) => {
|
||||
const prev = getEntity(entDef.name, id) as Record<string, unknown> | undefined
|
||||
const next = { ...(prev ?? {}), ...(patch as Record<string, unknown>) }
|
||||
setEntity(entDef.name, id, next)
|
||||
post.push({ type: entDef.name, id, patch: { op: Op.Merge, path: EMPTY_PATH, value: patch as Record<string, unknown> } })
|
||||
},
|
||||
removeEntity: (entDef, id) => {
|
||||
entityBucket(entDef.name).delete(id)
|
||||
post.push({ type: entDef.name, id, patch: { op: Op.Delete, path: EMPTY_PATH } })
|
||||
},
|
||||
upsertEntity: (entDef, entity) => {
|
||||
const id = entDef.id(entity)
|
||||
setEntity(entDef.name, id, entity)
|
||||
post.push({ type: entDef.name, id, patch: { op: Op.Set, path: EMPTY_PATH, value: entity } })
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function invalidate(def: MutationDef, input: unknown, resp: unknown): void {
|
||||
if (!def.invalidate) return
|
||||
const targets = def.invalidate(input, resp)
|
||||
for (const t of targets) {
|
||||
if (typeof t === 'string') {
|
||||
for (const node of nodes.values()) if (node.def.tags?.(node.args as never).includes(t)) void runFetch(node)
|
||||
} else {
|
||||
for (const node of nodes.values()) if (node.def === t) void runFetch(node)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const queue = createMutationQueue({
|
||||
storage,
|
||||
mutations: registry.mutations,
|
||||
emitEntityPatches,
|
||||
buildCtx,
|
||||
buildPostCtx,
|
||||
invalidate,
|
||||
isOnline,
|
||||
onOnline,
|
||||
onResult: (mutId, ok, data, error) =>
|
||||
endpoint.broadcast({ type: Msg.MutateResult, mutId, ok, data, error }),
|
||||
})
|
||||
|
||||
void queue.init()
|
||||
|
||||
endpoint.onClient((msg: ClientMsg) => {
|
||||
if (msg.type === Msg.Subscribe) void subscribe(msg)
|
||||
else if (msg.type === Msg.Unsubscribe) unsubscribe(msg.subId)
|
||||
else if (msg.type === Msg.Mutate) void queue.enqueue(msg.mutId, msg.defName, msg.input)
|
||||
else if (msg.type === Msg.FetchNextPage) fetchNextPage(msg.subId)
|
||||
})
|
||||
|
||||
return { nodes, subscribe, unsubscribe, fetchNextPage, queue }
|
||||
}
|
||||
|
||||
function shallowEqual(a: Record<string, unknown>, b: Record<string, unknown>): boolean {
|
||||
const ak = Object.keys(a)
|
||||
let bn = 0
|
||||
for (const _ in b) bn++
|
||||
if (ak.length !== bn) return false
|
||||
for (let i = 0; i < ak.length; i++) {
|
||||
const k = ak[i]
|
||||
if (a[k] !== b[k]) return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
function noop(): void {}
|
||||
Vendored
+12
@@ -0,0 +1,12 @@
|
||||
declare const __SYNC_ENGINE_DEV__: boolean
|
||||
|
||||
declare module 'virtual:sync-engine-registry' {
|
||||
import type { EntityDef, InfiniteQueryDef, MutationDef, QueryDef } from './engine/core/types'
|
||||
type AnyQueryDef = (QueryDef | InfiniteQueryDef) & { name: string }
|
||||
const registry: {
|
||||
entities: ReadonlyArray<EntityDef>
|
||||
queries: ReadonlyArray<AnyQueryDef>
|
||||
mutations: ReadonlyArray<MutationDef>
|
||||
}
|
||||
export default registry
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
import { createApp } from 'vue'
|
||||
import App from './App.vue'
|
||||
import { createTabEngine, createSharedWorkerClientTransport, installEngine } from './engine'
|
||||
|
||||
const worker = new SharedWorker(new URL('./engine.worker.ts', import.meta.url), {
|
||||
type: 'module',
|
||||
name: 'vue-sync-engine',
|
||||
})
|
||||
|
||||
const engine = createTabEngine({
|
||||
transport: createSharedWorkerClientTransport(worker),
|
||||
})
|
||||
|
||||
const app = createApp(App)
|
||||
installEngine(app, engine)
|
||||
app.mount('#app')
|
||||
@@ -0,0 +1,14 @@
|
||||
{
|
||||
"extends": "@vue/tsconfig/tsconfig.dom.json",
|
||||
"compilerOptions": {
|
||||
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.app.tsbuildinfo",
|
||||
"types": ["vite/client"],
|
||||
|
||||
/* Linting */
|
||||
"noUnusedLocals": true,
|
||||
"noUnusedParameters": true,
|
||||
"erasableSyntaxOnly": true,
|
||||
"noFallthroughCasesInSwitch": true
|
||||
},
|
||||
"include": ["src/**/*.ts", "src/**/*.tsx", "src/**/*.vue"]
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"files": [],
|
||||
"references": [
|
||||
{ "path": "./tsconfig.app.json" },
|
||||
{ "path": "./tsconfig.node.json" }
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"tsBuildInfoFile": "./node_modules/.tmp/tsconfig.node.tsbuildinfo",
|
||||
"target": "es2023",
|
||||
"lib": ["ES2023"],
|
||||
"module": "esnext",
|
||||
"types": ["node"],
|
||||
"skipLibCheck": true,
|
||||
|
||||
/* Bundler mode */
|
||||
"moduleResolution": "bundler",
|
||||
"allowImportingTsExtensions": true,
|
||||
"verbatimModuleSyntax": true,
|
||||
"moduleDetection": "force",
|
||||
"noEmit": true,
|
||||
|
||||
/* Linting */
|
||||
"noUnusedLocals": true,
|
||||
"noUnusedParameters": true,
|
||||
"erasableSyntaxOnly": true,
|
||||
"noFallthroughCasesInSwitch": true
|
||||
},
|
||||
"include": ["vite.config.ts"]
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
/// <reference types="vitest" />
|
||||
import { defineConfig } from "vite";
|
||||
import vue from "@vitejs/plugin-vue";
|
||||
import { syncEnginePlugin } from "./src/engine/plugin";
|
||||
|
||||
const enginePlugin = syncEnginePlugin({ definitions: ['/src/**/*.defs.ts'] });
|
||||
|
||||
export default defineConfig({
|
||||
plugins: [vue(), enginePlugin],
|
||||
worker: {
|
||||
plugins: () => [syncEnginePlugin({ definitions: ['/src/**/*.defs.ts'] })],
|
||||
},
|
||||
define: {
|
||||
__VUE_OPTIONS_API__: 'false',
|
||||
__VUE_PROD_DEVTOOLS__: 'false',
|
||||
__VUE_PROD_HYDRATION_MISMATCH_DETAILS__: 'false',
|
||||
__SYNC_ENGINE_DEV__: JSON.stringify(process.env.NODE_ENV !== 'production'),
|
||||
},
|
||||
test: {
|
||||
environment: "happy-dom",
|
||||
include: ["src/**/*.{test,spec}.ts"],
|
||||
globals: false,
|
||||
},
|
||||
});
|
||||
Reference in New Issue
Block a user