diff --git a/package-lock.json b/package-lock.json index 48eed99e..12f5cf4c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -15,6 +15,7 @@ "tg": "dist/cli.mjs" }, "devDependencies": { + "@cloudflare/vitest-pool-workers": "^0.18.6", "@electric-sql/pglite": "^0.5.4", "@eslint/js": "^10.0.1", "@standard-schema/spec": "^1.1.0", @@ -44,12 +45,16 @@ }, "peerDependencies": { "@electric-sql/pglite": "^0.4.4", + "better-sqlite3": "^12.11.1", "pg": "^8.20.0" }, "peerDependenciesMeta": { "@electric-sql/pglite": { "optional": true }, + "better-sqlite3": { + "optional": true + }, "pg": { "optional": true } @@ -452,6 +457,734 @@ } } }, + "node_modules/@cloudflare/vitest-pool-workers": { + "version": "0.18.6", + "resolved": "https://registry.npmjs.org/@cloudflare/vitest-pool-workers/-/vitest-pool-workers-0.18.6.tgz", + "integrity": "sha512-6JGqaQsQRZIVq/6jEC4ouJnShZriPIJ2X0yGndwMm+SiPP93pJi5Dp30dYAoztNrNJC7wWK7ec5slLfMBMZ8jA==", + "dev": true, + "license": "MIT", + "dependencies": { + "cjs-module-lexer": "1.2.3", + "esbuild": "0.28.1", + "miniflare": "4.20260714.0", + "wrangler": "4.112.0", + "zod": "3.25.76" + }, + "peerDependencies": { + "@vitest/runner": "^4.1.0", + "@vitest/snapshot": "^4.1.0", + "vitest": "^4.1.0" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@cloudflare/kv-asset-handler": { + "version": "0.5.0", + "resolved": "https://registry.npmjs.org/@cloudflare/kv-asset-handler/-/kv-asset-handler-0.5.0.tgz", + "integrity": "sha512-jxQYkj8dSIzc0cD6cMMNdOc1UVjqSqu8BZdor5s8cGjW2I8BjODt/kWPVdY+u9zj3ms75Q5qaZgnxUad83+eAg==", + "dev": true, + "license": "MIT OR Apache-2.0", + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@cloudflare/unenv-preset": { + "version": "2.16.1", + "resolved": "https://registry.npmjs.org/@cloudflare/unenv-preset/-/unenv-preset-2.16.1.tgz", + "integrity": "sha512-ECxObrMfyTl5bhQf/lZCXwo5G6xX9IAUo+nDMKK4SZ8m4Jvvxp52vilxyySSWh2YTZz8+HQ07qGH/2rEom1vDw==", + "dev": true, + "license": "MIT OR Apache-2.0", + "peerDependencies": { + "unenv": "2.0.0-rc.24", + "workerd": ">1.20260305.0 <2.0.0-0" + }, + "peerDependenciesMeta": { + "workerd": { + "optional": true + } + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@cloudflare/workerd-darwin-64": { + "version": "1.20260714.1", + "resolved": "https://registry.npmjs.org/@cloudflare/workerd-darwin-64/-/workerd-darwin-64-1.20260714.1.tgz", + "integrity": "sha512-ZWXqAN8G7Cx9hMRQuk+59ziJhR3j1F4iO+Qs8aHdfKZ3Dq5Yi/57xvkJTgCGBnW1YU/L78r8f6HEy51bwbTpNw==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "Apache-2.0", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@cloudflare/workerd-darwin-arm64": { + "version": "1.20260714.1", + "resolved": "https://registry.npmjs.org/@cloudflare/workerd-darwin-arm64/-/workerd-darwin-arm64-1.20260714.1.tgz", + "integrity": "sha512-tueWxWC3wyCbMG6zRAxsMXX0YLgrRWbiAPYFQ2uJ7dUH8G+5E7UTWaQS9B1HdJ0bpKFW1NWxhs1o2noKVFSUYg==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "Apache-2.0", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@cloudflare/workerd-linux-64": { + "version": "1.20260714.1", + "resolved": "https://registry.npmjs.org/@cloudflare/workerd-linux-64/-/workerd-linux-64-1.20260714.1.tgz", + "integrity": "sha512-1VChTZRb0l0F7R4e1G5RtLKV4oFi6x+rQgxh2+yu887j3l/3TLgatuv1L8/5zhc9gKEhATTxOh0e52Rtd9dDWQ==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "Apache-2.0", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@cloudflare/workerd-linux-arm64": { + "version": "1.20260714.1", + "resolved": "https://registry.npmjs.org/@cloudflare/workerd-linux-arm64/-/workerd-linux-arm64-1.20260714.1.tgz", + "integrity": "sha512-rMm3G+NirG2UdgHIRDdF1asNC6FqgIzZzkRG+VDhhDGcVxAQwvrMT1E38BivEvHr3G04MB4AfhcOczX0+GtRkQ==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "Apache-2.0", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@cloudflare/workerd-windows-64": { + "version": "1.20260714.1", + "resolved": "https://registry.npmjs.org/@cloudflare/workerd-windows-64/-/workerd-windows-64-1.20260714.1.tgz", + "integrity": "sha512-cGqnU3Hg2YZS/k3SAqrMp1DjpdsyFde72tWltdl6ZT9+SFz/Zrk/8gyTU1TcxC4YApXeNVH5TyU5cOGPgUJ0pg==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "Apache-2.0", + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=16" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@cloudflare/workers-types": { + "version": "5.20260719.1", + "resolved": "https://registry.npmjs.org/@cloudflare/workers-types/-/workers-types-5.20260719.1.tgz", + "integrity": "sha512-DcGasbfUuczQilc80vhL2MPPdWcaxWkx0hN5IW9UdNDBdrRvWcal3akSJ6Ccm7e8+/OGeR+8tYrqkykA3YGSZw==", + "dev": true, + "license": "MIT OR Apache-2.0", + "optional": true, + "peer": true + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/aix-ppc64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/aix-ppc64/-/aix-ppc64-0.28.1.tgz", + "integrity": "sha512-Svl7tq8k/08+p6CXPpRjQ1fKX+1odH/BQbb48fV6fj3CWHhsoIOoY87w1oHXm0qEpkIK3ZfVgp0hed3XBXzXMQ==", + "cpu": [ + "ppc64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "aix" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/android-arm": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/android-arm/-/android-arm-0.28.1.tgz", + "integrity": "sha512-0k2F129Xdio1TdJfzJ8sy1Q47vUD2NnwdhiAf7drUN1EBTfPf4hsFCtmMgu/6m8JSzsBrlmVjudMBQqOfG8usQ==", + "cpu": [ + "arm" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "android" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/android-arm64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/android-arm64/-/android-arm64-0.28.1.tgz", + "integrity": "sha512-34EGEbCIAgosYz6goLcopX6Mo7NyGv9tfwEM2/7Ce2VcVRk568iSvniGWcUXIy7wEDR1wzolcxcriFVrWYcwBg==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "android" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/android-x64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/android-x64/-/android-x64-0.28.1.tgz", + "integrity": "sha512-dbwY7ltSMDWsRatcRpCnES4F+im88OCUgGZjy52shC7GqHRE/cYlxNbB4Z4UpJswpcc4Qxd2oE/ufM0p61IKng==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "android" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/darwin-arm64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/darwin-arm64/-/darwin-arm64-0.28.1.tgz", + "integrity": "sha512-TZbWkQY7kvTAXbXUT7uVACR5cMHsDiSz9z7ZKAX/RTq/WJEk3QyRr0wZpNhBDX+/0CtdqUIJlOiodQcta6tY3Q==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/darwin-x64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/darwin-x64/-/darwin-x64-0.28.1.tgz", + "integrity": "sha512-zfdzgK9ACBNZLI/CyHTOx81SyNbM6YXn7rxSgX97VjyiPl9W1i4Ka4fgKECEoFCKGpvBj5qArWIGgQjOwkgskQ==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/freebsd-arm64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/freebsd-arm64/-/freebsd-arm64-0.28.1.tgz", + "integrity": "sha512-wG2EA8ENdEI0qhkSZMjfqrdY+ziCYCPMmtZjjIwOmXFjmyzEHn+UUxk5of+SYsjtfs3VpnlC7QLzSI5hY/rOAw==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "freebsd" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/freebsd-x64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/freebsd-x64/-/freebsd-x64-0.28.1.tgz", + "integrity": "sha512-i7dZ9vQgnvSCzi/rYCXNgtF/U+eKZNJBzu3eTQbRgHnM7tNSizLOkRFAl3qzVc/Op/u5YkHHa4pf/3DOYHthLQ==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "freebsd" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/linux-arm": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/linux-arm/-/linux-arm-0.28.1.tgz", + "integrity": "sha512-qVXBOHQS+d5Y722GwJzJUtOLlX7km3CraOaGormF1pDtPd2C/l1SHRPgjLunLGe51Sh5YYWKMFDyV4SxgMQYTQ==", + "cpu": [ + "arm" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/linux-arm64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/linux-arm64/-/linux-arm64-0.28.1.tgz", + "integrity": "sha512-yHs+0uc8+nvEAfAfxrWQKK5peSNzBc4PegcMO0EJ2hT71uA7vB8Ihg2e77R2P7SG5uYjPbHlLLmve4LLLRCf0g==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/linux-ia32": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/linux-ia32/-/linux-ia32-0.28.1.tgz", + "integrity": "sha512-d1z4ZuP0ajrfz/FhGT4vv278rX8KnPPJx8i5+AtK7TYbx9Le9F1hyzurZpkEyjkGa9dUGhQow4C1NmeGvqxN2w==", + "cpu": [ + "ia32" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/linux-loong64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/linux-loong64/-/linux-loong64-0.28.1.tgz", + "integrity": "sha512-M5sRjUVZrkm1OAPR3dlOYzNmN+loZKGVi1VUQGrwuqLcbR6qeAz+famMhjASeH3YVKvZz+zT1jlh/keC3Rj/lg==", + "cpu": [ + "loong64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/linux-mips64el": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/linux-mips64el/-/linux-mips64el-0.28.1.tgz", + "integrity": "sha512-mRObBZeHh2OxcBFPWE/FjylkRgZdYuiTR3vaTozquCGOH14iP9oN4x4Ge81CoIDYQrXmIxpFumJBu5MtZpnQJQ==", + "cpu": [ + "mips64el" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/linux-ppc64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/linux-ppc64/-/linux-ppc64-0.28.1.tgz", + "integrity": "sha512-slScBsMAb3GFDcdrCgLwZtPYRoH2H/youv10QiZyRjmsP48fznoveWytSgCI/R0ZcUgpc0ZhIUEx6LHts8yrfQ==", + "cpu": [ + "ppc64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/linux-riscv64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/linux-riscv64/-/linux-riscv64-0.28.1.tgz", + "integrity": "sha512-kw0owk1o0GFETUJyW0jc0G4Yzs0BHZn0JDZ8JRT088vjJYX777BAs1fDGxAC+q831qOs2DTC96mNsG2opdfyyQ==", + "cpu": [ + "riscv64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/linux-s390x": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/linux-s390x/-/linux-s390x-0.28.1.tgz", + "integrity": "sha512-/lAIjX8aYFRByhh6L5rYtPEDRqa9de/4V/juOXcta5frjvzXO4/sqEtyytse0g3zZFuWu5cDN0MkLz2qRDD2Ag==", + "cpu": [ + "s390x" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/linux-x64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/linux-x64/-/linux-x64-0.28.1.tgz", + "integrity": "sha512-u/anNYF2mmVOEDwLtnQ1wOr3EZ9sTNGLWrsYGYwHWzGA3Si84IOkHXlbWTD1NB+9/1lcnweYKO54uhxZydNzfA==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/netbsd-arm64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/netbsd-arm64/-/netbsd-arm64-0.28.1.tgz", + "integrity": "sha512-oks0DYbLwWMmaakTsCb+zL4E+aHRVLom9IJZOAthMQEPiQmydXHkziYEsGYRx0uNV/IjEKGAV941JzH02pflqw==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "netbsd" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/netbsd-x64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/netbsd-x64/-/netbsd-x64-0.28.1.tgz", + "integrity": "sha512-aeL6lAnN89Hz43Mlh1G8ARasbuoYvSITDEx0tHh5b7jJnHcssqgjy9Yx430GDpmCa6OyrKoS0aNRjKundRizGg==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "netbsd" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/openbsd-arm64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/openbsd-arm64/-/openbsd-arm64-0.28.1.tgz", + "integrity": "sha512-MEFJe5C3R8pwXdZ5Y21oo6m7ePiS0d9pWucn99O/wvyJZChoIQKrQDxKrGeW8F5+T0okTHesAmDeiHDTIq0V/Q==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "openbsd" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/openbsd-x64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/openbsd-x64/-/openbsd-x64-0.28.1.tgz", + "integrity": "sha512-i/ZLIOafE0Z8cI/XANJAixoJL/uRAoS2xOA3rb0xN+KK0K177cMAsQYkzHtBrtMXAKuAc7HGgcWiZ/sRC1Nxgw==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "openbsd" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/openharmony-arm64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/openharmony-arm64/-/openharmony-arm64-0.28.1.tgz", + "integrity": "sha512-ge+Z7EXFNt2BO1oAMsVpiQ8EwndV9i1xXerAeTIK7AtPs3bKFXQM7nlRxDSIUIMeueR1CNXxqztLzdNeReKBJg==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "openharmony" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/sunos-x64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/sunos-x64/-/sunos-x64-0.28.1.tgz", + "integrity": "sha512-BEjgtECkL3vY+SaSQ6nzVfiALUeFxpawyp8Jmf5PtYhf1Ug40N1h/hxlhts+f1FvSvarEigdxS3BlSMI2PJLcQ==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "sunos" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/win32-arm64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/win32-arm64/-/win32-arm64-0.28.1.tgz", + "integrity": "sha512-lCv9eK/H6ZJWbE7bh2nw54CZ9M2nupBxJcTsdk/QQnWkdSjKGuxmmH8/GWrlT1eMmZfn4dGcCjRte397WqfQXA==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/win32-ia32": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/win32-ia32/-/win32-ia32-0.28.1.tgz", + "integrity": "sha512-zvb/mB2bSCoJOpoCBgYKKpX6YM6mJBlBUVUtVj41DlZJVEB6/0CKlRYxP5wWl1C1ILiCoAU5wZZ4q1P3qeS6Eg==", + "cpu": [ + "ia32" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/@esbuild/win32-x64": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/@esbuild/win32-x64/-/win32-x64-0.28.1.tgz", + "integrity": "sha512-bm4Mowrv+GXMlpWX++EcXw/iLyd1o3+bJkC2DkWXYVvgZCqD/bSj9ctZeAMC3cIxgjRVR2Dufaiu4YPxr5gW1A==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT", + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=18" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/cjs-module-lexer": { + "version": "1.2.3", + "resolved": "https://registry.npmjs.org/cjs-module-lexer/-/cjs-module-lexer-1.2.3.tgz", + "integrity": "sha512-0TNiGstbQmCFwt4akjjBg5pLRTSyj/PkWQ1ZoO2zntmg9yLqSRxwEa4iCfQLGjqhiqBfOJa7W/E8wfGrTDmlZQ==", + "dev": true, + "license": "MIT" + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/esbuild": { + "version": "0.28.1", + "resolved": "https://registry.npmjs.org/esbuild/-/esbuild-0.28.1.tgz", + "integrity": "sha512-HrJrvZv5ayxBzPfwphOoNzkzOIIlifzk0KJrGK2c8R4+LKpMtpYLQeUdjnwjWv/LZlkH2laZk+4w78pi99D4Vw==", + "dev": true, + "hasInstallScript": true, + "license": "MIT", + "bin": { + "esbuild": "bin/esbuild" + }, + "engines": { + "node": ">=18" + }, + "optionalDependencies": { + "@esbuild/aix-ppc64": "0.28.1", + "@esbuild/android-arm": "0.28.1", + "@esbuild/android-arm64": "0.28.1", + "@esbuild/android-x64": "0.28.1", + "@esbuild/darwin-arm64": "0.28.1", + "@esbuild/darwin-x64": "0.28.1", + "@esbuild/freebsd-arm64": "0.28.1", + "@esbuild/freebsd-x64": "0.28.1", + "@esbuild/linux-arm": "0.28.1", + "@esbuild/linux-arm64": "0.28.1", + "@esbuild/linux-ia32": "0.28.1", + "@esbuild/linux-loong64": "0.28.1", + "@esbuild/linux-mips64el": "0.28.1", + "@esbuild/linux-ppc64": "0.28.1", + "@esbuild/linux-riscv64": "0.28.1", + "@esbuild/linux-s390x": "0.28.1", + "@esbuild/linux-x64": "0.28.1", + "@esbuild/netbsd-arm64": "0.28.1", + "@esbuild/netbsd-x64": "0.28.1", + "@esbuild/openbsd-arm64": "0.28.1", + "@esbuild/openbsd-x64": "0.28.1", + "@esbuild/openharmony-arm64": "0.28.1", + "@esbuild/sunos-x64": "0.28.1", + "@esbuild/win32-arm64": "0.28.1", + "@esbuild/win32-ia32": "0.28.1", + "@esbuild/win32-x64": "0.28.1" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/miniflare": { + "version": "4.20260714.0", + "resolved": "https://registry.npmjs.org/miniflare/-/miniflare-4.20260714.0.tgz", + "integrity": "sha512-MYlTCLdWCPqvrYY2uLwOjXwmglXuiHE3TGGkbOW4BwjUPa1r07E0iuHwrNDIs/sxK21r+o90Jx58AV2KeNdJZw==", + "dev": true, + "license": "MIT", + "dependencies": { + "@cspotcode/source-map-support": "0.8.1", + "sharp": "0.34.5", + "undici": "7.28.0", + "workerd": "1.20260714.1", + "ws": "8.21.0", + "youch": "4.1.0-beta.10" + }, + "bin": { + "miniflare": "bootstrap.js" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/undici": { + "version": "7.28.0", + "resolved": "https://registry.npmjs.org/undici/-/undici-7.28.0.tgz", + "integrity": "sha512-cRZYrTDwWznlnRiPjggAGxZXanty6M8RV1ff8Wm4LWXBp7/IG8v5DnOm74DtUBp9OONpK75YlPnIjQqX0dBDtA==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=20.18.1" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/workerd": { + "version": "1.20260714.1", + "resolved": "https://registry.npmjs.org/workerd/-/workerd-1.20260714.1.tgz", + "integrity": "sha512-oIbQzfdyl9UQUnG6XLegcSq0Mgt/7WKDbFOoqGgOWCS+/fhyGB460uKEgdAQQ9RHCO/ttcNCX/KiMIQzdoeu3Q==", + "dev": true, + "hasInstallScript": true, + "license": "Apache-2.0", + "bin": { + "workerd": "bin/workerd" + }, + "engines": { + "node": ">=16" + }, + "optionalDependencies": { + "@cloudflare/workerd-darwin-64": "1.20260714.1", + "@cloudflare/workerd-darwin-arm64": "1.20260714.1", + "@cloudflare/workerd-linux-64": "1.20260714.1", + "@cloudflare/workerd-linux-arm64": "1.20260714.1", + "@cloudflare/workerd-windows-64": "1.20260714.1" + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/wrangler": { + "version": "4.112.0", + "resolved": "https://registry.npmjs.org/wrangler/-/wrangler-4.112.0.tgz", + "integrity": "sha512-5H+XUD0TySCv1LuktFHDIEOkboH2nTfQs+35L+USt3MtntjDTMVIJprLgQcL2WBjulOyjxpd1vyTiSTJVW5MjQ==", + "dev": true, + "license": "MIT OR Apache-2.0", + "dependencies": { + "@cloudflare/kv-asset-handler": "0.5.0", + "@cloudflare/unenv-preset": "2.16.1", + "blake3-wasm": "2.1.5", + "esbuild": "0.28.1", + "miniflare": "4.20260714.0", + "path-to-regexp": "6.3.0", + "unenv": "2.0.0-rc.24", + "workerd": "1.20260714.1" + }, + "bin": { + "cf-wrangler": "bin/cf-wrangler.js", + "wrangler": "bin/wrangler.js", + "wrangler2": "bin/wrangler.js" + }, + "engines": { + "node": ">=22.0.0" + }, + "optionalDependencies": { + "fsevents": "2.3.3" + }, + "peerDependencies": { + "@cloudflare/workers-types": "^5.20260714.1" + }, + "peerDependenciesMeta": { + "@cloudflare/workers-types": { + "optional": true + } + } + }, + "node_modules/@cloudflare/vitest-pool-workers/node_modules/zod": { + "version": "3.25.76", + "resolved": "https://registry.npmjs.org/zod/-/zod-3.25.76.tgz", + "integrity": "sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ==", + "dev": true, + "license": "MIT", + "funding": { + "url": "https://github.com/sponsors/colinhacks" + } + }, "node_modules/@cloudflare/workerd-darwin-64": { "version": "1.20260310.1", "resolved": "https://registry.npmjs.org/@cloudflare/workerd-darwin-64/-/workerd-darwin-64-1.20260310.1.tgz", diff --git a/package.json b/package.json index 130e36d9..03a8bf77 100644 --- a/package.json +++ b/package.json @@ -58,8 +58,8 @@ ], "peerDependencies": { "@electric-sql/pglite": "^0.4.4", - "pg": "^8.20.0", - "better-sqlite3": "^12.11.1" + "better-sqlite3": "^12.11.1", + "pg": "^8.20.0" }, "peerDependenciesMeta": { "@electric-sql/pglite": { @@ -78,14 +78,16 @@ "scripts": { "build": "tsdown", "codegen": "node --experimental-strip-types src/types/postgres/emit.ts && node --experimental-strip-types src/types/sqlite/emit.ts", - "codegen:check": "tmp=$(mktemp -d) && node --experimental-strip-types src/types/postgres/emit.ts --out-dir \"$tmp/pg\" && { diff -r \"$tmp/pg\" src/types/postgres/generated || { echo 'src/types/postgres/generated is stale — run `npm run codegen` and commit.' >&2; rm -rf \"$tmp\"; exit 1; }; } && node --experimental-strip-types src/types/sqlite/emit.ts --out-dir \"$tmp/sqlite\" && { diff -r \"$tmp/sqlite/generated\" src/types/sqlite/generated || { echo 'src/types/sqlite/generated is stale — run `npm run codegen` and commit.' >&2; rm -rf \"$tmp\"; exit 1; }; } && rm -rf \"$tmp\"", + "codegen:check": "tmp=$(mktemp -d) && node --experimental-strip-types src/types/postgres/emit.ts --out-dir \"$tmp/pg\" && { diff -r \"$tmp/pg\" src/types/postgres/generated || { echo 'src/types/postgres/generated is stale - run `npm run codegen` and commit.' >&2; rm -rf \"$tmp\"; exit 1; }; } && node --experimental-strip-types src/types/sqlite/emit.ts --out-dir \"$tmp/sqlite\" && { diff -r \"$tmp/sqlite/generated\" src/types/sqlite/generated || { echo 'src/types/sqlite/generated is stale - run `npm run codegen` and commit.' >&2; rm -rf \"$tmp\"; exit 1; }; } && rm -rf \"$tmp\"", "lint": "eslint src", "typecheck": "tsgo --noEmit", "format": "prettier --write src", "test": "vitest run", - "check": "npm run lint && npm run typecheck && npm run test && npm --prefix examples/basic run check" + "check": "npm run lint && npm run typecheck && npm run test && npm --prefix examples/basic run check", + "test:do": "vitest run --project workerd" }, "devDependencies": { + "@cloudflare/vitest-pool-workers": "^0.18.6", "@electric-sql/pglite": "^0.5.4", "@eslint/js": "^10.0.1", "@standard-schema/spec": "^1.1.0", diff --git a/site/src/demo/runtime.ts b/site/src/demo/runtime.ts index 618b7304..5051b52b 100644 --- a/site/src/demo/runtime.ts +++ b/site/src/demo/runtime.ts @@ -2,7 +2,7 @@ // Uses top-level await so schema files can import a ready `db` and // define tables at module-eval time. -import { typegres } from "typegres"; +import { ensurePgLiveEventsTable, typegres } from "typegres"; import { runMigrations, runSeed } from "./seed"; import type { UserRoot } from "./server/api"; @@ -11,9 +11,8 @@ export const { db, conn } = await typegres({ type: "pglite" }); await runMigrations(conn); await runSeed(conn); -// `installLiveEvents` is a one-time DDL — production callers run it as +// `ensurePgLiveEventsTable` is a one-time DDL — production callers run it as // part of their migrations. The demo's storage doesn't survive page -// reloads, so we run it on every boot. `startLive` then assumes the -// events table exists and only spins up the polling bus. -await conn.installLiveEvents(); -await conn.startLive(); +// reloads, so we run it on every boot. The live engine itself is wired +// at attach and its poller starts lazily on first .live() use. +await ensurePgLiveEventsTable(conn); diff --git a/site/src/demo/schema/customers.ts b/site/src/demo/schema/customers.ts index 29dba474..f502bb06 100644 --- a/site/src/demo/schema/customers.ts +++ b/site/src/demo/schema/customers.ts @@ -1,9 +1,9 @@ -import { TypegresLiveEvents, expose } from "typegres"; +import { expose } from "typegres"; import { Int8, Text } from "typegres/postgres"; import { db } from "../runtime"; import { Orders } from "./orders"; import { Organizations } from "./organizations"; -export class Customers extends db.Table("customers", { transformer: TypegresLiveEvents.makeTransformer() }) { +export class Customers extends db.Table("customers", { live: true }) { // @generated-start @expose() id = (Int8<1>).column({ nonNull: true, generated: true }); @expose() organization_id = (Int8<1>).column({ nonNull: true }); diff --git a/site/src/demo/schema/inventory_positions.ts b/site/src/demo/schema/inventory_positions.ts index f6337691..12566ed2 100644 --- a/site/src/demo/schema/inventory_positions.ts +++ b/site/src/demo/schema/inventory_positions.ts @@ -1,11 +1,11 @@ -import { Connection, TypegresLiveEvents, sql, expose } from "typegres"; +import { Connection, sql, expose } from "typegres"; import { Int8, Text } from "typegres/postgres"; import { z } from "zod"; import { db } from "../runtime"; import { Locations } from "./locations"; import { Organizations } from "./organizations"; import { OrderLines } from "./order_lines"; -export class InventoryPositions extends db.Table("inventory_positions", { transformer: TypegresLiveEvents.makeTransformer() }) { +export class InventoryPositions extends db.Table("inventory_positions", { live: true }) { // @generated-start @expose() id = (Int8<1>).column({ nonNull: true, generated: true }); @expose() organization_id = (Int8<1>).column({ nonNull: true }); diff --git a/site/src/demo/schema/locations.ts b/site/src/demo/schema/locations.ts index 162bd694..5a93911d 100644 --- a/site/src/demo/schema/locations.ts +++ b/site/src/demo/schema/locations.ts @@ -1,10 +1,10 @@ -import { TypegresLiveEvents, expose } from "typegres"; +import { expose } from "typegres"; import { Int8, Text } from "typegres/postgres"; import { db } from "../runtime"; import { InventoryPositions } from "./inventory_positions"; import { Organizations } from "./organizations"; -export class Locations extends db.Table("locations", { transformer: TypegresLiveEvents.makeTransformer() }) { +export class Locations extends db.Table("locations", { live: true }) { // @generated-start @expose() id = (Int8<1>).column({ nonNull: true, generated: true }); @expose() organization_id = (Int8<1>).column({ nonNull: true }); diff --git a/site/src/demo/schema/order_lines.ts b/site/src/demo/schema/order_lines.ts index 296c0392..dbf8f268 100644 --- a/site/src/demo/schema/order_lines.ts +++ b/site/src/demo/schema/order_lines.ts @@ -1,10 +1,10 @@ -import { TypegresLiveEvents, expose } from "typegres"; +import { expose } from "typegres"; import { Int8, Text } from "typegres/postgres"; import { db } from "../runtime"; import { InventoryPositions } from "./inventory_positions"; import { Orders } from "./orders"; -export class OrderLines extends db.Table("order_lines", { transformer: TypegresLiveEvents.makeTransformer() }) { +export class OrderLines extends db.Table("order_lines", { live: true }) { // @generated-start @expose() id = (Int8<1>).column({ nonNull: true, generated: true }); @expose() order_id = (Int8<1>).column({ nonNull: true }); diff --git a/site/src/demo/schema/orders.ts b/site/src/demo/schema/orders.ts index 7290d393..01e507bf 100644 --- a/site/src/demo/schema/orders.ts +++ b/site/src/demo/schema/orders.ts @@ -1,4 +1,4 @@ -import { Connection, TypegresLiveEvents, sql, expose } from "typegres"; +import { Connection, sql, expose } from "typegres"; import { Int8, Text, Timestamptz } from "typegres/postgres"; import { z } from "zod"; import { db } from "../runtime"; @@ -7,7 +7,7 @@ import { OrderLines } from "./order_lines"; import { Organizations } from "./organizations"; import { Shipments } from "./shipments"; -export class Orders extends db.Table("orders", { transformer: TypegresLiveEvents.makeTransformer() }) { +export class Orders extends db.Table("orders", { live: true }) { // @generated-start @expose() id = (Int8<1>).column({ nonNull: true, generated: true }); @expose() organization_id = (Int8<1>).column({ nonNull: true }); @@ -51,9 +51,9 @@ export class Orders extends db.Table("orders", { transformer: TypegresLiveEvents // The WHERE excludes 'delivered' so CASE always matches; the // `as Text<1>` asserts the non-null we structurally guarantee. // Interpolating `orders.status` into the template emits the - // properly-qualified column reference (the live transformer - // wraps the UPDATE in a CTE that also has a `status` column; - // unqualified would be ambiguous). + // properly-qualified column reference (live capture wraps the + // UPDATE in a CTE that also has a `status` column; unqualified + // would be ambiguous). .set(({ orders }) => ({ status: Text.from(sql` CASE ${orders.status} diff --git a/site/src/demo/schema/organizations.ts b/site/src/demo/schema/organizations.ts index a654301b..23a2856e 100644 --- a/site/src/demo/schema/organizations.ts +++ b/site/src/demo/schema/organizations.ts @@ -1,4 +1,4 @@ -import { TypegresLiveEvents, expose } from "typegres"; +import { expose } from "typegres"; import { Int8, Text } from "typegres/postgres"; import { db } from "../runtime"; import { Customers } from "./customers"; @@ -7,7 +7,7 @@ import { Locations } from "./locations"; import { Users } from "./users"; import { Orders } from "./orders"; import { Shipments } from "./shipments"; -export class Organizations extends db.Table("organizations", { transformer: TypegresLiveEvents.makeTransformer() }) { +export class Organizations extends db.Table("organizations", { live: true }) { // @generated-start @expose() id = (Int8<1>).column({ nonNull: true, generated: true }); @expose() name = (Text<1>).column({ nonNull: true }); diff --git a/site/src/demo/schema/shipments.ts b/site/src/demo/schema/shipments.ts index 0790512b..0ae2af4d 100644 --- a/site/src/demo/schema/shipments.ts +++ b/site/src/demo/schema/shipments.ts @@ -1,9 +1,9 @@ -import { TypegresLiveEvents, sql, expose } from "typegres"; +import { sql, expose } from "typegres"; import { Int8, Text, Timestamptz } from "typegres/postgres"; import { db } from "../runtime"; import { Orders } from "./orders"; import { Organizations } from "./organizations"; -export class Shipments extends db.Table("shipments", { transformer: TypegresLiveEvents.makeTransformer() }) { +export class Shipments extends db.Table("shipments", { live: true }) { // @generated-start @expose() id = (Int8<1>).column({ nonNull: true, generated: true }); @expose() organization_id = (Int8<1>).column({ nonNull: true }); diff --git a/site/src/demo/schema/users.ts b/site/src/demo/schema/users.ts index e2ad6f7f..7c9c1321 100644 --- a/site/src/demo/schema/users.ts +++ b/site/src/demo/schema/users.ts @@ -1,8 +1,8 @@ -import { TypegresLiveEvents, expose } from "typegres"; +import { expose } from "typegres"; import { Int8, Text } from "typegres/postgres"; import { db } from "../runtime"; import { Organizations } from "./organizations"; -export class Users extends db.Table("users", { transformer: TypegresLiveEvents.makeTransformer() }) { +export class Users extends db.Table("users", { live: true }) { // @generated-start @expose() id = (Int8<1>).column({ nonNull: true, generated: true }); @expose() organization_id = (Int8<1>).column({ nonNull: true }); diff --git a/site/src/demo/server/api.ts b/site/src/demo/server/api.ts index 8ccd4446..c5d0ea21 100644 --- a/site/src/demo/server/api.ts +++ b/site/src/demo/server/api.ts @@ -148,16 +148,14 @@ export class Api { this.#currentUserToken = token; } - // Demo stop-button hook. `db.stopLive()` cancels every active - // subscription (parked consumers wake with AbortError and exit - // cleanly), then we re-start the bus so the next watch can - // subscribe immediately. In a real deployment the wire would have - // a per-iter abort channel; here we tear down the whole bus - // because the demo only ever has one iter at a time. + // Demo stop-button hook: cancels every active subscription (parked + // consumers wake with AbortError and exit cleanly); the live engine + // stays up and the next watch subscribes immediately. In a real + // deployment the wire would have a per-iter abort channel; here we + // cancel all subs because the demo only ever has one iter at a time. @expose() async resetLive(): Promise { - await this.conn.stopLive(); - await this.conn.startLive(); + this.conn.cancelLiveSubscriptions(); } // The principal for this RPC, resolved from the ambient diff --git a/src/builder/delete.ts b/src/builder/delete.ts index 6dd6f904..f8b1f66c 100644 --- a/src/builder/delete.ts +++ b/src/builder/delete.ts @@ -18,8 +18,8 @@ type DeleteOpts = { }; // Resolved DELETE — alias minted, where/returning evaluated against the -// bound namespace. Transformers inspect structured fields without -// re-parsing raw Sql. +// bound namespace. Consumers (live capture) inspect structured fields +// without re-parsing raw Sql. type FinalizedDeleteOpts = { tableName: Name; alias: Alias; @@ -59,12 +59,10 @@ export class DeleteBuilder); + return this.#opts.returning?.({ [this.table.tableName]: this.#opts.instance } as Namespace); } finalize(): FinalizedDelete { if (!this.#opts.where && !this.#opts.matchAll) { throw new Error("delete() requires .where() — use .where(true) to delete all rows"); } - const tableName = this.tableName; + const tableName = this.table.tableName as Name; const alias = new Alias(tableName); const instance = reAlias(this.#opts.instance as RowType, alias) as T; const ns = { [tableName]: instance } as Namespace; @@ -120,8 +118,10 @@ export class DeleteBuilder((v) => isRowType(v)))) @@ -127,11 +125,11 @@ export class InsertBuilder); + return this.#opts.returning?.({ [this.table.tableName]: this.#opts.instance } as Namespace); } finalize(): FinalizedInsert { - const tableName = this.tableName; + const tableName = this.table.tableName as Name; const alias = new Alias(tableName); const instance = reAlias(this.#opts.instance as RowType, alias) as T; const ns = { [tableName]: instance } as Namespace; @@ -147,8 +145,10 @@ export class InsertBuilder( fn: (ns: Namespace) => R2, ): UpdateBuilder { @@ -131,7 +129,7 @@ export class UpdateBuilder); + return this.#opts.returning?.({ [this.table.tableName]: this.#opts.instance } as Namespace); } finalize(): FinalizedUpdate { @@ -141,7 +139,7 @@ export class UpdateBuilder; @@ -160,8 +158,10 @@ export class UpdateBuilder { // Attach a driver → get a runtime Connection. Multiple `attach` calls // are allowed (test + prod, worker pools, replicas) — Connections // share the schema provenance but talk to independent drivers. - attach(driver: Driver): Connection { + // + // The live engine is wired here too: sqlite capture is active from the + // start; the pg poller spins up lazily on first .live() use. `liveOpts` + // configures the pg bus (poll cadence, backfill window). + attach(driver: Driver, liveOpts?: BusOptions): Connection { if (driver.dialect !== this.dialect) { throw new Error( `Driver dialect '${driver.dialect}' does not match Database dialect '${this.dialect}'.`, ); } - return new Connection(this, driver); + return new Connection(this, driver, undefined, undefined, liveOpts); } // Entry point for non-Table Fromables (SRFs, Values, subqueries). @@ -109,9 +113,16 @@ export class Database { // Runtime handle: has a driver, executes queries. Constructed via // `db.attach(driver)`. `.transaction()` mints a txn-bound Connection -// sharing the same driver + Database. `.close()` closes the driver. +// sharing the same driver + Database. `.close()` stops the live bus and +// closes the driver. export class Connection { - #boundExecute?: ExecuteFn; + // The execution context: one dialect executor, wired at construction; + // transaction() mints the bound variant. + #executor: Executor; + // Pool-backed connections own a live bus (sqlite: fed synchronously by + // capture; pg: polls once started). Tx-bound connections share their + // parent's via the executor closures and hold none themselves. + #bus: Bus | undefined; // Active isolation on a txn-bound Connection. `undefined` means either // pool-backed (no active txn) or ambient (txn opened without an // explicit level — we deferred to pg's session default, which we @@ -121,20 +132,32 @@ export class Connection { constructor( readonly database: Database, private driver: Driver, - boundExecute?: ExecuteFn, + executor?: Executor, isolation?: TransactionIsolation, + liveOpts?: BusOptions, ) { - if (boundExecute) { this.#boundExecute = boundExecute; } + if (executor) { + // Transaction-bound: share the parent's context; no bus of our own. + this.#executor = executor; + } else if (database.dialect === "postgres") { + this.#bus = new Bus(this, liveOpts); + this.#executor = new PgExecutor(database, (compiled) => driver.execute(compiled)); + } else { + if (!isSyncDriver(driver)) { + throw new Error( + "sqlite requires a SyncDriver (SqliteDriver, DoSqliteDriver) — " + + "live capture must run pre-image read + mutation + dispatch without awaits between them", + ); + } + const bus = new Bus(this, liveOpts); + this.#bus = bus; + this.#executor = new SqliteLiveExecutor(database, driver, bus); + } if (isolation) { this.#isolation = isolation; } } get dialect() { return this.driver.dialect; } - #exec(query: Sql): Promise { - const compiled = compile(query, { database: this.database }); - return (this.#boundExecute ?? this.driver.execute.bind(this.driver))(compiled); - } - // Overload resolution matches top-to-bottom and stops on the first // match — list specific builders before the general `Sql` fallback, or // every QueryBuilder/Insert/... call resolves as `Sql` → QueryResult. @@ -152,7 +175,7 @@ export class Connection { ): Promise ? RowTypeToTsType[] : never>; async execute(query: Sql): Promise; async execute(query: Sql): Promise { - const result = await this.#exec(query); + const result = await this.#executor.run(query); if (query instanceof QueryBuilder) { return deserializeRows(result.rows as { [key: string]: string }[], query.rowType()); } @@ -181,7 +204,7 @@ export class Connection { query: DeleteBuilder, ): Promise; async hydrate(query: Sql): Promise { - const result = await this.#exec(query); + const result = await this.#executor.run(query); let shape: { [k: string]: unknown } | undefined; if (query instanceof QueryBuilder) { shape = query.rowType() as { [k: string]: unknown }; @@ -206,8 +229,17 @@ export class Connection { if (!fn) { throw new Error("transaction() requires a callback"); } - if (this.#boundExecute) { + if (opts?.isolation && this.database.dialect !== "postgres") { + // Sqlite rejects pg's BEGIN ISOLATION LEVEL syntax, and its + // transactions are serializable by nature — no level to pick. + throw new Error( + `transaction({ isolation }) is pg-only — sqlite transactions are always serializable; omit the option`, + ); + } + if (this.#executor.bound) { // Already in a txn — flatten, but reject silent isolation downgrades. + // (Flattening shares the executor, so sqlite's event buffer spans + // to the outermost commit.) if (opts?.isolation) { const active = this.#isolation; if (active === undefined) { @@ -225,13 +257,46 @@ export class Connection { } return fn(this); } + const bus = this.#bus!; return this.driver.runInSingleConnection(async (execute) => { - const tx = new Connection(this.database, this.driver, execute, opts?.isolation); + const driver = this.driver; + let txExecutor: Executor; + if (this.database.dialect === "postgres") { + txExecutor = new PgExecutor(this.database, execute, true); + } else if (isSyncDriver(driver)) { + // Bound and pooled are the same channel on sqlite's one handle — + // checked, not assumed; the bound executor differs only in event + // timing (commit-deferred flush). + if (execute !== driver.executeSync) { + throw new Error( + "sync driver must pass its executeSync to runInSingleConnection — one handle, one channel", + ); + } + txExecutor = new SqliteLiveExecutor(this.database, driver, bus, true); + } else { + // The constructor rejects async sqlite drivers at attach — this + // branch exists so TS narrows `driver` above. + throw new Error("unreachable: sqlite Connection without a SyncDriver"); + } + const tx = new Connection(this.database, this.driver, txExecutor, opts?.isolation); + // Drivers with a native transaction protocol (Durable Objects) own + // commit/rollback; everyone else gets BEGIN/COMMIT/ROLLBACK SQL. + if (driver.runInTransaction) { + try { + const result = await driver.runInTransaction(() => fn(tx)); + txExecutor.onCommit(); + return result; + } catch (e) { + txExecutor.onRollback(); + throw e; + } + } const runSql = async (s: Sql) => execute(compile(s, { database: this.database })); await runSql(opts?.isolation ? ISOLATION[opts.isolation].begin : sql`BEGIN`); try { const result = await fn(tx); await runSql(sql`COMMIT`); + txExecutor.onCommit(); return result; } catch (e) { try { @@ -239,40 +304,27 @@ export class Connection { } catch (rollbackErr) { console.error("ROLLBACK failed after transaction error:", rollbackErr); } + txExecutor.onRollback(); throw e; } }); } async close(): Promise { - if (this.#boundExecute) { + if (this.#executor.bound) { throw new Error("close() must be called on a pool-backed Connection, not inside a transaction"); } + await this.#bus?.stop(); await this.driver.close(); } // --- Live queries --- - #bus: Bus | undefined; - - async installLiveEvents(): Promise { - for (const stmt of eventsTableSqlStatements(this.database)) { - await this.driver.execute(compile(stmt, { database: this.database })); - } - } - - async startLive(opts: BusOptions = {}): Promise { - if (this.#boundExecute) { - throw new Error("startLive() must be called on a pool-backed Connection, not inside a transaction"); - } - if (this.#bus) { throw new Error("Live bus already started"); } - this.#bus = new Bus(this, opts); - await this.#bus.start(); - } - async stopLive(): Promise { - const bus = this.#bus; - this.#bus = undefined; - await bus?.stop(); + // Cancel every active live subscription (parked consumers wake with + // AbortError and their generators end cleanly). The engine stays up — + // the next .live() call subscribes as usual. + cancelLiveSubscriptions(): void { + this.#bus?.cancelSubscriptions(); } async *live>( @@ -282,17 +334,28 @@ export class Connection { ? RowTypeToTsType[] : never > { - if (this.#boundExecute) { + if (this.#executor.bound) { throw new Error("live() can't be called inside a transaction"); } - const bus = this.#bus; - if (!bus) { throw new Error("Live bus not started — call conn.startLive() first"); } + const bus = this.#bus!; + // Lazy engine start: a no-op on sqlite (capture feeds the bus + // synchronously from attach); on pg this seeds the snapshot watermark + // and spins the poll loop on first use, so connections that never + // call .live() never poll. + await bus.ensureStarted(); let currentSub: Subscription | undefined; try { while (true) { - const { rows, cursor, predicateSet } = await runLiveIteration(this, query); - currentSub = bus.subscribe(parseSnapshot(cursor), predicateSet); + const epoch = bus.epoch; + const { rows, cursor, predicateSet } = await this.#executor.runLiveIteration(this, query); + // A cancel (or stop) that landed while the iteration ran has no + // parked Subscription to reject for us — honor it here instead + // of re-subscribing past it. + if (bus.epoch !== epoch) { + return; + } + currentSub = bus.subscribe(cursor, predicateSet); yield rows as any; if (currentSub) { try { diff --git a/src/drivers/do.ts b/src/drivers/do.ts index b24b6ed9..951eae54 100644 --- a/src/drivers/do.ts +++ b/src/drivers/do.ts @@ -1,5 +1,5 @@ import type { CompiledSql } from "../builder/sql"; -import type { Driver, ExecuteFn, QueryResult } from "./types"; +import type { ExecuteFn, ExecuteSyncFn, QueryResult, SyncDriver } from "./types"; import { normalizeRow, stripMatchedOuterParens } from "./shared-sqlite"; // Duck-typed Cloudflare SqlStorage — no @cloudflare/workers-types dependency. @@ -7,7 +7,16 @@ export interface SqlStorageLike { exec(query: string, ...bindings: unknown[]): { toArray(): { [key: string]: unknown }[] }; } -// typegres Driver over a Durable Object's SQLite (Cloudflare SqlStorage). +// Duck-typed Cloudflare DurableObjectStorage (the parts we need). The +// whole storage handle, not just `.sql`: workerd rejects SQL BEGIN/ +// SAVEPOINT on SqlStorage — transactions must use storage.transaction(). +export interface DoStorageLike { + readonly sql: SqlStorageLike; + transaction(cb: () => Promise): Promise; +} + +// typegres Driver over a Durable Object's SQLite storage: +// const conn = db.attach(new DoSqliteDriver(ctx.storage)); // No node-only peer imports — safe to bundle into workerd. // // Contract notes: @@ -16,19 +25,33 @@ export interface SqlStorageLike { // callers / the dialect must not emit bigint bindings for this backend; // - results are native JS values, normalized to strings for deserialize; // - blobs come back as ArrayBuffer (better-sqlite3 gives Uint8Array/Buffer). -export class DoSqliteDriver implements Driver { +export class DoSqliteDriver implements SyncDriver { readonly dialect = "sqlite" as const; - constructor(private readonly sql: SqlStorageLike) {} + #liveSeq = 0n; + get liveSeq(): bigint { + return this.#liveSeq; + } + + constructor(private readonly storage: DoStorageLike) {} - execute: ExecuteFn = ({ text, values }: CompiledSql): Promise => { + execute: ExecuteFn = (compiled: CompiledSql): Promise => + Promise.resolve(this.executeSync(compiled)); + + executeSync = ({ text, values }: CompiledSql): QueryResult => { + this.#liveSeq++; const query = stripMatchedOuterParens(text); - const rows = this.sql.exec(query, ...values).toArray().map(normalizeRow); - return Promise.resolve({ rows }); + const rows = this.storage.sql.exec(query, ...values).toArray().map(normalizeRow); + return { rows }; }; - runInSingleConnection = (cb: (execute: ExecuteFn) => Promise): Promise => - cb(this.execute); + // storage.transaction() commits on resolution, rolls back on throw. + runInTransaction = (cb: () => Promise): Promise => this.storage.transaction(cb); + + // One handle: the single-connection execute IS executeSync (callers + // assert this identity — see Connection.transaction). + runInSingleConnection = (cb: (execute: ExecuteSyncFn) => Promise): Promise => + cb(this.executeSync); close = (): Promise => Promise.resolve(); } diff --git a/src/drivers/sqlite.ts b/src/drivers/sqlite.ts index c525a34a..d9b0dc92 100644 --- a/src/drivers/sqlite.ts +++ b/src/drivers/sqlite.ts @@ -1,15 +1,20 @@ import type { CompiledSql } from "../builder/sql"; import type { DialectName } from "../builder/sql"; import type BetterSqlite3 from "better-sqlite3"; -import type { Driver, ExecuteFn, QueryResult } from "./types"; +import type { ExecuteSyncFn, QueryResult, SyncDriver } from "./types"; import { normalizeRow, stripMatchedOuterParens } from "./shared-sqlite"; // better-sqlite3 adapter. Synchronous under the hood; wrapped in // Promise.resolve for the async Driver contract. `better-sqlite3` is an // optional peer (see package.json). -export class SqliteDriver implements Driver { +export class SqliteDriver implements SyncDriver { readonly dialect: DialectName = "sqlite"; + #liveSeq = 0n; + get liveSeq(): bigint { + return this.#liveSeq; + } + static async create( filename: string = ":memory:", options: BetterSqlite3.Options = {}, @@ -22,11 +27,16 @@ export class SqliteDriver implements Driver { private constructor(private db: BetterSqlite3.Database) {} - async execute({ text, values }: CompiledSql): Promise { + async execute(compiled: CompiledSql): Promise { + return this.executeSync(compiled); + } + + executeSync = ({ text, values }: CompiledSql): QueryResult => { + this.#liveSeq++; // QueryBuilder.bind() wraps statements in `(...)` for subquery splicing; // SQLite refuses top-level parenthesized statements — unwrap one matched pair. return this.runOne(stripMatchedOuterParens(text), values); - } + }; private runOne(text: string, values: readonly unknown[]): QueryResult { const stmt = this.db.prepare(text); @@ -40,8 +50,10 @@ export class SqliteDriver implements Driver { return { rows: [] }; } - async runInSingleConnection(cb: (execute: ExecuteFn) => Promise): Promise { - return cb((compiled) => Promise.resolve(this.execute(compiled))); + async runInSingleConnection(cb: (execute: ExecuteSyncFn) => Promise): Promise { + // One handle: the single-connection execute IS executeSync (callers + // assert this identity — see Connection.transaction). + return cb(this.executeSync); } async close(): Promise { diff --git a/src/drivers/types.ts b/src/drivers/types.ts index 927ec225..b84538ef 100644 --- a/src/drivers/types.ts +++ b/src/drivers/types.ts @@ -14,9 +14,35 @@ export type QueryResult = { rows: { [key: string]: string | null }[] }; // hand the query to its underlying pool/wasm and normalize the result rows. export type ExecuteFn = (sql: CompiledSql) => Promise; +export type ExecuteSyncFn = (sql: CompiledSql) => QueryResult; +// Callers `await` either flavor (a no-op on the sync one). +export type AnyExecuteFn = ExecuteFn | ExecuteSyncFn; + export interface Driver { readonly dialect: DialectName; execute: ExecuteFn; - runInSingleConnection(cb: (execute: ExecuteFn) => Promise): Promise; + // Present when the engine is sync under the hood (better-sqlite3, DO + // SqlStorage). Required by sqlite live capture, which needs multiple + // statements with no awaits between them. + executeSync?: ExecuteSyncFn; + // Native transaction protocol: commit when `cb` resolves, roll back + // when it throws. When present, Connection.transaction() uses this + // instead of BEGIN/COMMIT/ROLLBACK SQL (workerd rejects SQL BEGIN). + runInTransaction?(cb: () => Promise): Promise; + // Sync drivers must pass their executeSync itself as `execute` (one + // handle, one channel) — Connection.transaction() asserts the identity. + runInSingleConnection(cb: (execute: AnyExecuteFn) => Promise): Promise; close(): Promise; } + +// A driver that guarantees the synchronous path; sqlite live requires one. +export interface SyncDriver extends Driver { + executeSync: ExecuteSyncFn; + // Statement clock, incremented internally per executed statement — the + // sqlite analog of pg's xids. Live capture stamps events with it; live + // iterations read it as their cursor. Shared by all Connections on + // this driver. + readonly liveSeq: bigint; +} + +export const isSyncDriver = (d: Driver): d is SyncDriver => d.executeSync !== undefined; diff --git a/src/executor.ts b/src/executor.ts new file mode 100644 index 00000000..821f8bdc --- /dev/null +++ b/src/executor.ts @@ -0,0 +1,83 @@ +import { compile, type CompiledSql, type Sql } from "./builder/sql"; +import type { Connection, Database } from "./database"; +import type { QueryResult } from "./drivers/types"; +import type { QueryBuilder, RowType, RowTypeToTsType } from "./builder/query"; +import { InsertBuilder } from "./builder/insert"; +import { UpdateBuilder } from "./builder/update"; +import { DeleteBuilder } from "./builder/delete"; +import type { PredicateSet } from "./live/extractor"; +import type { Cursor } from "./live/snapshot"; + +export type MutationOp = "insert" | "update" | "delete"; +export type MutationBuilder = + | InsertBuilder + | UpdateBuilder + | DeleteBuilder; + +const mutationOp = (query: Sql): MutationOp | undefined => + query instanceof InsertBuilder ? "insert" + : query instanceof UpdateBuilder ? "update" + : query instanceof DeleteBuilder ? "delete" + : undefined; + +// Has the mutation's target table opted into live capture (`{ live: true }`)? +export const isLiveMutation = (builder: MutationBuilder): boolean => builder.table.live; + +export type LiveIterationResult = { + rows: RowTypeToTsType[]; + // Ready to hand straight to bus.subscribe(). pg: the parsed + // pg_current_snapshot(); sqlite: the driver's seq embedded in the same + // Cursor shape. + cursor: Cursor; + // Closed set of watched values per (table, col) — propagated across + // equality classes. + predicateSet: PredicateSet; +}; + +// A Connection's execution context: one concrete executor per dialect +// (live/pg/executor.ts, live/sqlite/executor.ts), constructed in pooled +// or transaction-bound mode. run() detects mutation builders and routes +// them to runMutation (the per-dialect capture hook); everything bottoms +// out in runStatement (compile + exec). +export abstract class Executor { + constructor( + readonly database: Database, + // True for transaction-bound executors — drives Connection's "not + // inside a transaction" guards and transaction() flattening. + readonly bound: boolean = false, + ) {} + + protected abstract exec(compiled: CompiledSql): Promise; + + run(query: Sql): Promise { + const op = mutationOp(query); + return op ? this.runMutation(op, query as MutationBuilder) : this.runStatement(query); + } + + protected runStatement(query: Sql): Promise { + return this.exec(compile(query, { database: this.database })); + } + + // Default: mutations are just statements. Dialect executors override + // with their capture behavior. + protected runMutation(_op: MutationOp, builder: MutationBuilder): Promise { + return this.runStatement(builder); + } + + // One iteration of the live loop: capture a cursor "before" the query's + // view of the data, compute the watched predicate set, run the query. + // Takes the Connection because pg wraps this in conn.transaction(). + abstract runLiveIteration>( + conn: Connection, + query: Q, + ): Promise< + Q extends QueryBuilder + ? LiveIterationResult + : never + >; + + // Fired by Connection.transaction() after COMMIT/ROLLBACK. No-ops + // unless a variant holds deferred work (sqlite's event buffer). + onCommit(): void {} + onRollback(): void {} +} diff --git a/src/index.ts b/src/index.ts index 27852908..cf5253af 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,13 +11,14 @@ export { Table } from "./table"; export { Relation } from "./relation"; export { sql, Sql } from "./builder/sql"; export { QueryBuilder } from "./builder/query"; -export { TypegresLiveEvents } from "./live/events"; +export { TypegresLiveEvents } from "./live/pg/events"; +export { ensurePgLiveEventsTable } from "./live/pg/events-ddl"; export { expose } from "./exoeval/tool"; export type { ToolFunction } from "./exoeval/tool"; export { RpcClient, inMemoryChannel, safeStringify } from "./exoeval/rpc"; export type { RawChannel } from "./exoeval/rpc"; export type { Config } from "./config"; -export type { Driver, ExecuteFn, QueryResult } from "./drivers/types"; +export type { Driver, SyncDriver, ExecuteFn, ExecuteSyncFn, QueryResult } from "./drivers/types"; import type { Connection } from "./database"; import { Database } from "./database"; diff --git a/src/live/ISSUES.md b/src/live/ISSUES.md index 61ce4490..6cf37f68 100644 --- a/src/live/ISSUES.md +++ b/src/live/ISSUES.md @@ -11,32 +11,32 @@ A query is live-able iff **every table reference has at least one top-level Otherwise arbitrary: `JOIN`, `LEFT JOIN`, `GROUP BY`, `HAVING`, subqueries — fine. The extractor only inspects top-level `WHERE`/`ON`. -## Open - -1. **Public API gap: live mode isn't actually consumable from the package yet.** - The npm entrypoint exports `Database`, `db.startLive()`, and `db.live()`, - but not the setup pieces users need to make live mode work: - - `TypegresLiveEvents.makeTransformer()` to opt tables into event capture - - `TypegresLiveEvents.createTableSql()` to create `_typegres_live_events` +The Bus (index, subscribe/backfill, lifecycle) is dialect-shared; only +the **event source** differs. pg (`pg/`): shadow events table + polling + +MVCC snapshot cursors. sqlite (`sqlite/`): no shadow table and no polling +— the runtime is a synchronous single writer (better-sqlite3, a Durable +Object's SqlStorage), so RETURNING-image capture pushes events into +`Bus.ingest()` in the same tick as the mutation (buffered until COMMIT +inside transactions — see SqliteLiveExecutor), entirely in memory; +cursors are integer seqs embedded in the snapshot `Cursor` shape so the +MVCC `visible()` test doubles as the seq comparison. The scope rule above +applies to both dialects. - `package.json` also uses explicit export maps, so consumers can't rely on - a deep import escape hatch. Result: the live subsystem exists, but from a - package user's perspective it's effectively private / unusable. - - Fix options: export `TypegresLiveEvents`; expose a higher-level - `enableLive()` helper; or intentionally keep the whole feature internal - until the setup surface is ready. +## Open -2. **`stopLive()` leaves active subscriptions wedged.** - Shutting down the bus stops polling, but today it does not drain existing - subscriptions or settle their `wait` promises. Any async iterator parked in - `await currentSub.wait` can hang forever if `db.stopLive()` is called while - it's waiting. The old Bus instance also keeps stale subscription/index state - alive until GC. +1. **Public API gap, closed.** Opt-in is uniformly + `db.Table("notes", { live: true })`; the live engine is wired at + `db.attach(driver, busOpts?)` with no start/stop lifecycle — sqlite + capture is active from attach, the pg poller starts lazily on first + `.live()` use, and `close()` tears the engine down. The only pg + ceremony left is `ensurePgLiveEventsTable(conn)` for the events-table + DDL (a migration concern, deliberately explicit). - `Bus.stop()` should explicitly resolve/reject/unsubscribe outstanding subs, - and live iterators should observe shutdown as completion or error rather - than hanging. +2. ~~**`stopLive()` leaves active subscriptions wedged.**~~ Resolved: + `Bus.stop()` cancels outstanding subscriptions (rejects parked `wait`s + with AbortError; the live generator observes shutdown as clean + completion). Covered by the sqlite + cancelLiveSubscriptions / close() teardown tests. 3. **More `.live()` tests.** Five today (insert/update/delete/join/not-started). Missing: backfill-from-buffer (subscribe returns `undefined`, caller @@ -47,14 +47,14 @@ subqueries — fine. The extractor only inspects top-level `WHERE`/`ON`. while an iterator is waiting. 4. **Circular-import workarounds.** - - `bus.ts` inlines `const EVENTS_TABLE = "_typegres_live_events"` - because importing `TypegresLiveEvents` would cycle through - `events.ts → table.ts → insert.ts → database.ts → bus.ts`, and - `events.ts`'s `class extends Table(...)` evaluates `Table` at - module-load (TDZ). - - Fix options: hoist the constant to a Table-free module; lazy class - factory; restructure deps so the live modules don't pull `database.ts`. + - `bus.ts` inlines `const EVENTS_TABLE = "_typegres_live_events"` to + keep events.ts out of its import chain. Historical note: the + original TDZ crash came from an events.ts that did `class extends + Table(...)` at module load; the current events.ts is class-free re + Table, and it now sits in database's load graph via executor.ts → + live/pg/events.ts with no incident (verified against the site vite + build, the original crash site). The bus's inlined constant is + belt-and-suspenders at this point. 5. **No `pg_notify` / `LISTEN`** — rerun latency floor is the poll interval (default 100ms). Adding `pg_notify('typegres_live_events', '')` @@ -82,21 +82,59 @@ subqueries — fine. The extractor only inspects top-level `WHERE`/`ON`. perf footnote: an atomic-swap that diffs old vs. new and only mutates leaves that changed would be cheaper for stable predicate sets. -9. **Text canonicalization landmines** — both extractor and events - transformer canonicalize values via `col::text`; bus matches by string - equality. Symmetric *as long as both sides agree on text form*. Three - pg types drift: - - **`timestamptz`** — `::text` honors session `TimeZone`; same instant, - different text in different sessions. - - **`citext`** — case-insensitive at compare time but `::text` preserves - case; equal values can produce non-equal text. - - **`numeric`** — trailing zeros / scale survive `::text`; - `1.0::numeric` and `1::numeric` compare equal but differ as text. - - No live test exercises any of these today; a user query against one of - these column types may silently miss matches. Fix range: canonicalizing - casts (`lower()` / `extract(epoch from …)`) up to a binary comparator - in the bus. +9. **Text canonicalization landmines** — the matcher replays SQL `=` + *outside* SQL: it lifts predicate values out of the query and compares + canonical text in JS. That loses every normalization `=` would have + applied (operator resolution, pg implicit casts, sqlite affinity / + numeric cross-class comparison), so the invariant the whole scheme + rests on is: **the canonical rendering must be constant on each SQL + equality class** — two `=`-equal values that render differently are a + silently *missed wakeup* (always a false negative; conflating unequal + values merely causes a harmless spurious rerun). + + `::text` / `CAST(x AS text)` satisfies this for the common types + (int, text, uuid, bool, float8) when both sides render through the + same type's output function. Known drift: + + **pg** (pre-existing, unfixed): + - **`numeric`** — scale survives `::text`: `1 = 1.0 = 1.00` but + `'1'` / `'1.0'` / `'1.00'` (verified). `trim_scale()` (pg13+) + normalizes. This is the most likely real-world miss. + - **`timestamptz`** — `::text` honors session `TimeZone`/`DateStyle`; + event images render in the *writer's* session, extractor values in + the live connection's session — differing GUCs break matching. + Same-pool same-settings holds in practice but is undocumented. + - **`citext`** / non-binary collations — equal at compare time, + case-preserving as text. + + **sqlite** (as of the sqlite backend): + - **Fixed**: drivers bind every JS number as REAL while + integer-affinity columns store INTEGER (`'1.0'` vs `'1'`); + `canonical.ts` collapses integral REALs before the text cast. + Root cause: sqlite operators live on root `Any`, which binds bare + `?` params (`sql-value.ts` — `CAST(? AS any)` would be wrong), so + unlike pg's `TypedParam` nothing re-normalizes the binding class. + A parity test hangs-on-timeout if the render sites ever diverge. + - **Open (contrived)**: TEXT-affinity column vs number param — + SQL applies TEXT affinity to REAL `1.0` → matches stored `'1.0'`, + but the collapsed anchor renders `'1'` → miss. Reachable because + root-`Any.eq` accepts any primitive. + - **Harmless**: no-affinity columns distinguish integer `1` from + text `'1'`; canonical text conflates them → spurious rerun only. + + Fix directions (complementary, roughly in order of leverage): + 1. **Typed casts in the sqlite operator codegen** — per-type arg slots + so `eq` emits `CAST(? AS INTEGER)` etc., mimicking compare-time + affinity at the binding layer. Fixes the anchor at the source + (including the TEXT-affinity edge) and would let `canonical.ts`'s + CASE collapse be deleted. Correct independent of live. + 2. **Per-type canonical-render hook** on the type classes (all three + render sites hold typed values): `Numeric → trim_scale(x)::text`, + `timestamptz → UTC-normalized`, sqlite classes → their affinity + semantics (or JS-side `String()` canonicalization, which collapses + `1.0`/`1` for free). Subsumes `canonical.ts` and fixes pg numeric. + 3. Status quo: `canonical.ts`'s dialect-level rendering — works for + the common types, locked by tests, ad-hoc. 10. **Unsupported predicate forms produce a generic "unrooted" error** instead of a clear "this predicate form isn't supported" message — @@ -135,7 +173,10 @@ subqueries — fine. The extractor only inspects top-level `WHERE`/`ON`. correctly qualified. 12. **WAL / logical-replication ingestion as an alternative to the shadow - table** — today every mutation gains a `_typegres_live_events` insert + table** (pg; the sqlite backend already has no shadow table — its + analog of this gap is that RETURNING-image capture also only sees + builder-routed writes, where per-table triggers would catch raw SQL + too) — today every pg mutation gains a `_typegres_live_events` insert in the same statement. That captures only writes that go through typegres builders, costs an extra row per mutation, and grows a table the user has to manage. A WAL-backed mode (logical decoding via a @@ -144,3 +185,84 @@ subqueries — fine. The extractor only inspects top-level `WHERE`/`ON`. replication permissions, per-table `REPLICA IDENTITY` config, slot monitoring (a stalled consumer holds WAL forever). Worth offering as a swappable backend for clients who prefer it. + +13. **Cap'n Web streaming surface (deferred; learnings recorded).** + `.live()` is a local AsyncIterable on both dialects (already + wire-consumable via exoeval's streaming); a capnweb push surface — + `subscribe(conn, cb)` driving the generator with `await cb(rows)` + backpressure — was prototyped and works, deferred to keep the live + core minimal. What the prototype established, so the next attempt + starts from knowledge: + - capnweb's `map` recorder serializes any captured plain function as a + nested record-replay closure (async ones are rejected outright: + "RPC closures cannot be async functions"). Passing a callback **by + reference** requires a plain RPC call — build the refined query + capability via `doRpc`, then call `subscribe` directly on the stub. + - An argument stub is auto-released when the call frame returns + ("RpcImportHook was already disposed" on the first push). The callee + must `dup()` the callback on entry and dispose the retained handle + on unsubscribe — which is exactly the pin/unpin-a-Durable-Object + lifecycle, and what the harness leak guard verifies. + - A driver loop that always has a `next()` in flight parks the live + generator at `await sub.wait` — a non-yield suspension `.return()` + cannot interrupt. `Connection.live()` needs an AbortSignal (or + equivalent) that `cancel()`s the in-flight Subscription so + unsubscribe can wake a parked consumer. + - Client-side, capnweb invokes an exported callback with args as + RpcPromises — the client cb must resolve them. + +14. **Multi-connection sqlite live / `LiveDriver` (parked).** Two + Connections attached to one sqlite driver share the statement clock + (`SyncDriver.liveSeq`) but NOT a bus — each attach() makes its + own, and events captured through one Connection never reach the + other's subscribers. Unexercised today (a DO is one driver, one + Connection). The named future shape: a user-constructed `LiveDriver` + wrapper (`db.attach(new LiveDriver(new SqliteDriver(...)))`) owning + the clock AND a per-driver bus, reverting the base drivers to dumb + pipes. Notes from the design discussion: a per-statement driver + callback cannot absorb capture (CompiledSql has no table/column/ + builder context — interception must stay at the Connection layer), + and per-statement clock ticking is not load-bearing (only capture- + time advancement matters for cursor correctness). + +15. **sqlite shared-handle interleaving (both drivers; rare; fix deferred).** + Sqlite has ONE handle/session, so a task that runs concurrently with an + open transaction's await gaps executes its SQL INSIDE that transaction: + a live rerun's SELECTs see uncommitted (possibly later-rolled-back) + rows, and a pooled-connection mutation is silently absorbed into the + transaction while its event was already ingested — a phantom event if + the transaction rolls back. Narrow in practice: commit-deferred flush + means live signals never originate mid-transaction, so hitting it + requires app code running its own concurrency (Promise.all etc.) + against an open transaction. + + **Durable Objects ARE affected too** — empirically verified in workerd + (probe since deleted): the input gate only blocks OTHER EVENTS (new + requests, alarms, timers) while `storage.transaction()` is open; + same-invocation microtask siblings interleave freely at the callback's + await points, read uncommitted state, and their writes join/roll back + with the transaction. Two corollaries: (a) a DO tx callback must only + await storage ops — awaiting anything resolved by sibling code + deadlocks (the gated sibling never runs... unless it's microtask-only, + in which case it interleaves instead); (b) on the DO driver a + mid-transaction cursor also never self-heals, because the native + commit path runs no driver statement — `liveSeq` isn't bumped at + COMMIT, so the transaction's events stamp with the last in-txn + statement's seq, which such a cursor already considers seen. + + Deferred fix (sketch): make the sqlite live iteration a synchronous + frame (compile + executeSync + deserializeRows — kills the + extractor/user-query straddle), park iterations on a Connection-held + open-transaction promise before entering the frame, and bump (or + stamp past) the statement clock at commit on the native-transaction + path. + +16. **Mutation builders embedded in raw sql templates bypass capture.** + Capture fires only when the builder is the top-level argument to + execute/hydrate (Executor.run's instanceof dispatch). A live-table + mutation interpolated into a sql`` template compiles as a plain + statement — no events, silently stale subscribers. (On main, pg's + bind-time transformer covered simple template embedding; the + CTE-embedded form was invalid there too.) No in-repo code uses the + pattern; fix would be detection at compile/bind time or documenting + the boundary alongside the raw-SQL one in item 12. diff --git a/src/live/bus.test.ts b/src/live/bus.test.ts index 41caa24b..44c17c5a 100644 --- a/src/live/bus.test.ts +++ b/src/live/bus.test.ts @@ -3,7 +3,7 @@ import { sql } from "../builder/sql"; import { conn, setupDb } from "../test-helpers"; import { Bus, CursorTooOldError, type Subscription } from "./bus"; import { type Cursor, parseSnapshot } from "./snapshot"; -import { setupLiveEvents } from "./test-helpers"; +import { setupLiveEvents } from "./pg/test-helpers"; setupDb(); setupLiveEvents(); @@ -35,7 +35,7 @@ const waitedWithin = async (sub: Subscription, ms = 50): Promise => test("Bus signals a subscription whose cursor doesn't see a matching event", async () => { const bus = new Bus(conn); - await bus.start(); + await bus.ensureStarted(); const sub = bus.subscribe( await grabSnapshot(), @@ -53,7 +53,7 @@ test("Bus signals a subscription whose cursor doesn't see a matching event", asy test("Bus does not signal when the cursor already sees the event", async () => { const bus = new Bus(conn); - await bus.start(); + await bus.ensureStarted(); // Insert event FIRST, then capture cursor — sub's snapshot sees it. await insertEvent("users", null, { id: "5", name: "Rex" }); @@ -71,7 +71,7 @@ test("Bus does not signal when the cursor already sees the event", async () => { test("subscribe returns undefined when in-memory backfill already shows a matching event", async () => { const bus = new Bus(conn); - await bus.start(); + await bus.ensureStarted(); // Capture cursor BEFORE the event commits (sub won't see it). const subCursor = await grabSnapshot(); @@ -98,7 +98,7 @@ test("subscribe returns undefined when in-memory backfill already shows a matchi test("subscribe throws CursorTooOldError when cursor is older than the buffer floor", async () => { // Tiny window of 2 — easy to roll the floor past an old cursor. const bus = new Bus(conn, { windowSize: 2 }); - await bus.start(); + await bus.ensureStarted(); // Snapshot held by an "ancient" cursor predating any committed events. const oldCursor = await grabSnapshot(); diff --git a/src/live/bus.ts b/src/live/bus.ts index 45644053..90545a9f 100644 --- a/src/live/bus.ts +++ b/src/live/bus.ts @@ -62,19 +62,18 @@ export class CursorTooOldError extends Error { } } -// Parsed once at ingest in #poll so the matcher can iterate plain arrays -// without re-parsing JSON or re-allocating BigInts on every event/sub -// lookup. -type EventRow = { +// `xid` is a pg transaction id or a caller-stamped sqlite seq — the bus +// only ever compares it through `visible()`. +export type EventRow = { xid: bigint; table: string; - // (col, value::text) pairs from event.before ∪ event.after — the shape - // both jsonb columns share. Null sides contribute nothing. + // (col, canonical-text value) pairs from event.before ∪ event.after — + // the shape both image sides share. Null sides contribute nothing. pairs: [string, string][]; }; export type BusOptions = { - // Polling cadence in ms. Default 100. + // Polling cadence in ms (pg only). Default 100. intervalMs?: number; // In-memory backfill window size. Default 10_000. The bus retains the // most-recent N events; subscriptions whose cursor predates the floor @@ -82,6 +81,14 @@ export type BusOptions = { windowSize?: number; }; +// The bus is dialect-shared; only its EVENT SOURCE differs: +// - pg: a polling loop reads `_typegres_live_events` and dispatches +// events newly visible since the last snapshot watermark. +// - sqlite: no shadow table, no polling — capture pushes events into +// `ingest()` in the mutation's own tick, with seq-counter xids and +// seq-embedded cursors (seqCursor) so the MVCC `visible()` test +// doubles as the seq comparison. +// Index, subscribe/backfill, and lifecycle are identical either way. export class Bus { // Reverse index: table → col → value → set of subscriptions. #index: Map>>> = new Map(); @@ -89,6 +96,7 @@ export class Bus { // Bus's processed-through cursor: every poll computes "events newly // visible since #watermark", then advances #watermark to the new // snapshot. Without it we'd re-dispatch already-handled events. + // (pg only — sqlite events arrive exactly once via ingest.) #watermark: Cursor | undefined; // Last N events observed by the bus, oldest-first. Used for sub // backfill scans without a DB roundtrip. @@ -118,17 +126,25 @@ export class Bus { this.#database = conn.database; } - // Capture the initial snapshot and start the polling loop. Must be - // called before subscribe(). - async start(): Promise { - if (this.#running) { - throw new Error("Bus already started"); - } + // Lazy start, awaited by each live() call before it subscribes — pg + // connections that never use .live() never poll. The cached promise + // (not a boolean) is what makes concurrent callers WAIT for the + // watermark seed: a subscriber slipping in mid-seed would pass the + // floor check vacuously and could miss events older than the seed. + #startPromise: Promise | undefined; + ensureStarted(): Promise { + return (this.#startPromise ??= this.#start()); + } + + // Capture the initial snapshot and start the polling loop (pg). On + // sqlite there is nothing to seed or poll — events arrive via ingest(). + async #start(): Promise { this.#running = true; - // Seed: bus's watermark is "now"; floor is now's xmin (no events - // older than this can be backfilled, but no events older than this - // would need to be — they're committed-and-visible to anyone with a - // cursor.xmin ≥ this). + if (this.#database.dialect !== "postgres") { + return; + } + // Seed: bus's watermark is "now"; floor is now's xmin (nothing older + // needs backfill — it's already visible to any cursor taken later). this.#watermark = await this.#readCursor(); this.#floor = this.#watermark.xmin; this.#startLoop(); @@ -141,19 +157,46 @@ export class Bus { this.#loopPromise = undefined; this.#watermark = undefined; this.#buffer = []; - // Wake any consumers parked on `await sub.wait` so their generators - // can run finally + return cleanly. cancel() rejects `wait`; the - // live iterator catches AbortError and exits. - for (const sub of [...this.#subs]) {sub.cancel();} + this.cancelSubscriptions(); + } + + // Cancellation generation. Bumped by cancelSubscriptions(); the live() + // loop snapshots it per iteration and exits if it changed — that + // covers iterators that were MID-RERUN at cancel time (not parked, so + // no Subscription to reject) which would otherwise re-subscribe and + // outlive the cancel. + #epoch = 0; + get epoch(): number { + return this.#epoch; + } + + // Wake every consumer parked on `await sub.wait` (cancel() rejects it; + // the live iterator catches AbortError and exits cleanly). Used by + // stop() and Connection.cancelLiveSubscriptions. + cancelSubscriptions(): void { + this.#epoch++; + for (const sub of [...this.#subs]) { + sub.cancel(); + } + } + + // Push-side event source (sqlite): called synchronously right after a + // mutation, with caller-stamped monotonic xids — the single-writer + // replacement for the pg poll. + ingest(events: readonly EventRow[]): void { + if (!this.#running || events.length === 0) { + return; + } + this.#dispatch([...events]); } // Returns undefined if the in-memory backfill buffer already shows a // mutation the cursor doesn't see — caller should rerun immediately, // no need to wait on anything. Otherwise returns a Subscription whose - // `wait` resolves on the next matching poll. + // `wait` resolves on the next matching event. subscribe(cursor: Cursor, predicateSet: PredicateSet): Subscription | undefined { if (!this.#running) { - throw new Error("Bus not started — call db.startLive() first"); + throw new Error("Bus not started — live() awaits ensureStarted() before subscribing"); } if (cursor.xmin < this.#floor) { throw new CursorTooOldError(cursor.xmin, this.#floor); @@ -175,7 +218,7 @@ export class Bus { } } - // No backfill match — index for future polls. + // No backfill match — index for future events. const { promise: wait, resolve, reject } = Promise.withResolvers(); // wait gets attached to one of resolve/reject in the call below; // before that, swallow unhandled-rejection if cancel fires before @@ -250,11 +293,15 @@ export class Bus { } // Force one poll iteration — for tests so they don't race against - // the timer. Resolves after the next poll cycle completes. + // the timer. Resolves after the next poll cycle completes. On sqlite + // dispatch is synchronous, so there is no loop to force — no-op. async pollNow(): Promise { if (!this.#running) { throw new Error("Bus not started"); } + if (!this.#loopPromise) { + return; + } const done = new Promise((resolve) => { this.#oncePolled.push(resolve); }); @@ -316,11 +363,19 @@ export class Bus { before: string | null; after: string | null; }[]; - const newEvents: EventRow[] = raw.map((row) => ({ - xid: BigInt(row.xid), - table: row.table, - pairs: parseEventPairs(row.before, row.after), - })); + this.#dispatch( + raw.map((row) => ({ + xid: BigInt(row.xid), + table: row.table, + pairs: parseEventPairs(row.before, row.after), + })), + ); + } + + // Shared tail of both event sources (pg #poll, sqlite ingest): signal + // matching subs whose cursor doesn't already see the event, then append + // to the in-memory window and trim. + #dispatch(newEvents: EventRow[]): void { for (const event of newEvents) { for (const sub of this.#subsMatching(event)) { if (!visible(sub.cursor, event.xid)) { @@ -362,10 +417,10 @@ export class Bus { } // Flatten event.before/after (each `{col: "stringified value", ...}` — -// same canonicalization as the extractor's `jsonb_build_object('col', -// col::text, ...)`) into a single array of (col, value) pairs. Run once -// per event at ingest so per-sub matching never re-parses JSON. -const parseEventPairs = (before: string | null, after: string | null): [string, string][] => { +// same canonicalization as the extractor's text-cast values) into a +// single array of (col, value) pairs. Run once per event at ingest so +// per-sub matching never re-parses JSON. +export const parseEventPairs = (before: string | null, after: string | null): [string, string][] => { const out: [string, string][] = []; for (const raw of [before, after]) { if (!raw) { diff --git a/src/live/canonical.ts b/src/live/canonical.ts new file mode 100644 index 00000000..406bdaff --- /dev/null +++ b/src/live/canonical.ts @@ -0,0 +1,20 @@ +import { Cast, type DialectName, sql, type Sql } from "../builder/sql"; + +// Canonical text rendering for live predicate matching — the contract +// shared by the extractor's UNION and each dialect's change capture (pg +// jsonb images via the equivalent `::text`; sqlite json_object images). +// All sites for one dialect must agree, or the matcher's string +// comparison silently misses. +// +// pg: plain CAST(x AS text) — typed params already bind at the column's +// canonical type. +// +// sqlite: drivers bind every JS number as REAL while integer-affinity +// columns store INTEGER, so a naive cast renders the anchor `1` as '1.0' +// but the stored value as '1'. Collapse integral reals to INTEGER before +// the text cast; non-integral reals ('1.5') and other storage classes +// pass through. (CAST(x AS numeric) can't do this — it's a no-op on REAL.) +export const canonicalText = (expr: Sql, dialect: DialectName): Sql => + dialect === "sqlite" + ? sql`CASE WHEN typeof(${expr}) = 'real' AND ${expr} = CAST(${expr} AS integer) THEN CAST(CAST(${expr} AS integer) AS text) ELSE CAST(${expr} AS text) END` + : new Cast(expr, sql`text`, dialect); diff --git a/src/live/extractor.test.ts b/src/live/extractor.test.ts index 5e72c63a..b5fabc16 100644 --- a/src/live/extractor.test.ts +++ b/src/live/extractor.test.ts @@ -84,10 +84,10 @@ test("buildExtractor: single-table literal anchor", () => { expectSqlEqual( buildExtractor(sortAliases(traverse(q))), sql` - WITH "users" AS MATERIALIZED ( - SELECT "id" FROM "users" AS "users" WHERE ("users"."id" = CAST(${"5"} AS int8)) + WITH "users_2" AS MATERIALIZED ( + SELECT "id" FROM "users" AS "users_2" WHERE ("users_2"."id" = CAST(${"5"} AS int8)) ) - SELECT ${"users"} AS tbl, ${"id"} AS col, CAST(${"5"} AS int8)::text AS value + SELECT ${"users"} AS tbl, ${"id"} AS col, CAST(CAST(${"5"} AS int8) AS text) AS value `, db, ); @@ -104,19 +104,19 @@ test("buildExtractor: self-join produces two CTEs both backed by 'users'", () => buildExtractor(sortAliases(traverse(q))), sql` WITH - "users" AS MATERIALIZED ( - SELECT "manager_id", "id" FROM "users" AS "users" - WHERE ("users"."id" = CAST(${"5"} AS int8)) + "users_2" AS MATERIALIZED ( + SELECT "manager_id", "id" FROM "users" AS "users_2" + WHERE ("users_2"."id" = CAST(${"5"} AS int8)) ), "manager" AS MATERIALIZED ( SELECT "id" FROM "users" AS "manager" - WHERE "manager"."id" IN (SELECT "manager_id" FROM "users") + WHERE "manager"."id" IN (SELECT "manager_id" FROM "users_2") ) - SELECT ${"users"} AS tbl, ${"manager_id"} AS col, "users"."manager_id"::text AS value FROM "users" + SELECT ${"users"} AS tbl, ${"manager_id"} AS col, CAST("users_2"."manager_id" AS text) AS value FROM "users_2" UNION ALL - SELECT ${"users"} AS tbl, ${"id"} AS col, CAST(${"5"} AS int8)::text AS value + SELECT ${"users"} AS tbl, ${"id"} AS col, CAST(CAST(${"5"} AS int8) AS text) AS value UNION ALL - SELECT ${"users"} AS tbl, ${"id"} AS col, "manager"."id"::text AS value FROM "manager" + SELECT ${"users"} AS tbl, ${"id"} AS col, CAST("manager"."id" AS text) AS value FROM "manager" `, db, ); @@ -131,18 +131,18 @@ test("buildExtractor: join chain rewrites edge to IN (SELECT … FROM upstream C buildExtractor(sortAliases(traverse(q))), sql` WITH - "users" AS MATERIALIZED ( - SELECT "id" FROM "users" AS "users" WHERE ("users"."id" = CAST(${"5"} AS int8)) + "users_2" AS MATERIALIZED ( + SELECT "id" FROM "users" AS "users_2" WHERE ("users_2"."id" = CAST(${"5"} AS int8)) ), - "dogs" AS MATERIALIZED ( - SELECT "user_id" FROM "dogs" AS "dogs" - WHERE "dogs"."user_id" IN (SELECT "id" FROM "users") + "dogs_2" AS MATERIALIZED ( + SELECT "user_id" FROM "dogs" AS "dogs_2" + WHERE "dogs_2"."user_id" IN (SELECT "id" FROM "users_2") ) - SELECT ${"users"} AS tbl, ${"id"} AS col, "users"."id"::text AS value FROM "users" + SELECT ${"users"} AS tbl, ${"id"} AS col, CAST("users_2"."id" AS text) AS value FROM "users_2" UNION ALL - SELECT ${"users"} AS tbl, ${"id"} AS col, CAST(${"5"} AS int8)::text AS value + SELECT ${"users"} AS tbl, ${"id"} AS col, CAST(CAST(${"5"} AS int8) AS text) AS value UNION ALL - SELECT ${"dogs"} AS tbl, ${"user_id"} AS col, "dogs"."user_id"::text AS value FROM "dogs" + SELECT ${"dogs"} AS tbl, ${"user_id"} AS col, CAST("dogs_2"."user_id" AS text) AS value FROM "dogs_2" `, db, ); @@ -159,21 +159,21 @@ test("buildExtractor: multiple anchors keep their literal predicates plus join e buildExtractor(sortAliases(traverse(q))), sql` WITH - "users" AS MATERIALIZED ( - SELECT "id" FROM "users" AS "users" WHERE ("users"."id" = CAST(${"5"} AS int8)) + "users_2" AS MATERIALIZED ( + SELECT "id" FROM "users" AS "users_2" WHERE ("users_2"."id" = CAST(${"5"} AS int8)) ), - "dogs" AS MATERIALIZED ( - SELECT "user_id", "name" FROM "dogs" AS "dogs" - WHERE "dogs"."user_id" IN (SELECT "id" FROM "users") - AND ("dogs"."name" = CAST(${"Rex"} AS text)) + "dogs_2" AS MATERIALIZED ( + SELECT "user_id", "name" FROM "dogs" AS "dogs_2" + WHERE "dogs_2"."user_id" IN (SELECT "id" FROM "users_2") + AND ("dogs_2"."name" = CAST(${"Rex"} AS text)) ) - SELECT ${"users"} AS tbl, ${"id"} AS col, "users"."id"::text AS value FROM "users" + SELECT ${"users"} AS tbl, ${"id"} AS col, CAST("users_2"."id" AS text) AS value FROM "users_2" UNION ALL - SELECT ${"users"} AS tbl, ${"id"} AS col, CAST(${"5"} AS int8)::text AS value + SELECT ${"users"} AS tbl, ${"id"} AS col, CAST(CAST(${"5"} AS int8) AS text) AS value UNION ALL - SELECT ${"dogs"} AS tbl, ${"user_id"} AS col, "dogs"."user_id"::text AS value FROM "dogs" + SELECT ${"dogs"} AS tbl, ${"user_id"} AS col, CAST("dogs_2"."user_id" AS text) AS value FROM "dogs_2" UNION ALL - SELECT ${"dogs"} AS tbl, ${"name"} AS col, CAST(${"Rex"} AS text)::text AS value + SELECT ${"dogs"} AS tbl, ${"name"} AS col, CAST(CAST(${"Rex"} AS text) AS text) AS value `, db, ); diff --git a/src/live/extractor.ts b/src/live/extractor.ts index aa93b0c1..7a627acc 100644 --- a/src/live/extractor.ts +++ b/src/live/extractor.ts @@ -1,22 +1,29 @@ import type { Connection } from "../database"; -import { FinalizedQuery, type QueryBuilder, type RowType, type RowTypeToTsType } from "../builder/query"; -import { type Alias, Column, Op, type Raw, type Sql, sql, TypedParam } from "../builder/sql"; +import { FinalizedQuery, type QueryBuilder } from "../builder/query"; +import { Alias, Column, Op, Param, type Raw, type Sql, sql, TypedParam } from "../builder/sql"; +import { canonicalText } from "./canonical"; import type { Database } from "../database"; import { type TableBase, isTableClass } from "../table"; type Table = typeof TableBase; +// A literal side of an equality predicate. Pg's typed wrappers bind +// primitives as `CAST($n AS T)` (TypedParam); sqlite's runtime binds them +// as bare `?` (Param). Both re-bind cleanly in the extractor's UNION. +type LiteralParam = TypedParam | Param; +const isLiteralParam = (s: Sql): s is LiteralParam => s instanceof TypedParam || s instanceof Param; + type EqualityPredicate = Op & { op: Raw & { value: "=" }; - lhs: Column | TypedParam; - rhs: Column | TypedParam; + lhs: Column | LiteralParam; + rhs: Column | LiteralParam; }; const extractPredicates = (root: Sql): EqualityPredicate[] => { // Extract top-level AND'ed equality predicates where at least one side - // is a Column. The other side is either another Column (→ edge) or any - // other Sql shape — typically `CAST($n AS T)` for typegres-wrapped - // primitives — which we treat as a literal anchor. + // is a Column. The other side is either another Column (→ edge) or a + // bound literal — `CAST($n AS T)` for pg-wrapped primitives, bare `?` + // for sqlite — which we treat as a literal anchor. const predicates: EqualityPredicate[] = []; const stack = [root]; @@ -30,7 +37,7 @@ const extractPredicates = (root: Sql): EqualityPredicate[] => { stack.push(node.rhs); } if (node instanceof Op && node.op.value === "=") { - const validSide = (s: Sql) => s instanceof Column || s instanceof TypedParam; + const validSide = (s: Sql) => s instanceof Column || isLiteralParam(s); const isCol = (s: Sql) => s instanceof Column; if (validSide(node.lhs) && validSide(node.rhs) && (isCol(node.lhs) || isCol(node.rhs))) { predicates.push(node as EqualityPredicate); @@ -42,7 +49,7 @@ const extractPredicates = (root: Sql): EqualityPredicate[] => { type RelativePredicate = { col: Column; // my column - to: Column | TypedParam; // other column or literal I'm equal to + to: Column | LiteralParam; // other column or literal I'm equal to // The original Op — handy when emitting WHERE clauses (already // parenthesizes itself on bind), so we don't have to reconstruct // `col = to` from the destructured sides. @@ -140,7 +147,7 @@ export type CteSpec = { export const sortAliases = (traversal: TraverseResult): CteSpec[] => { // Topo: seed with literal-anchored aliases, then walk edges until stable. const order: Alias[] = [...traversal.entries()] - .filter(([, entry]) => entry.predicates.some((p) => p.to instanceof TypedParam)) + .filter(([, entry]) => entry.predicates.some((p) => isLiteralParam(p.to))) .map(([alias]) => alias); const orderSet = new Set(order); @@ -192,7 +199,7 @@ export const buildExtractor = (specs: CteSpec[]): Sql => { const whereClauses: Sql[] = []; for (const p of spec.predicates) { - if (p.to instanceof TypedParam) { + if (isLiteralParam(p.to)) { // Anchor: reuse the original Op whereClauses.push(p.original); } else if (p.to instanceof Column) { @@ -205,7 +212,7 @@ export const buildExtractor = (specs: CteSpec[]): Sql => { // The upstream alias is also the upstream CTE's name — refer to it directly. whereClauses.push(sql`${p.col} IN (SELECT ${p.to.name} FROM ${p.to.tableAlias})`); } else { - throw new Error(`Unexpected non-Column, non-TypedParam predicate side: ${p.to}`); + throw new Error(`Unexpected non-Column, non-literal predicate side: ${p.to}`); } } @@ -223,15 +230,27 @@ export const buildExtractor = (specs: CteSpec[]): Sql => { }); const unionParts: Sql[] = specs.flatMap((spec) => { + // canonicalText is the value-rendering contract shared with each + // backend's change capture (pg jsonb images, sqlite json_object + // images) — see live/canonical.ts. + const asText = (expr: Sql) => canonicalText(expr, spec.database.dialect); return spec.predicates.map((pred) => - pred.to instanceof TypedParam - ? sql`SELECT ${sql.param(spec.tableName)} AS tbl, ${sql.param(pred.col.name.name)} AS col, ${pred.to}::text AS value` - : sql`SELECT ${sql.param(spec.tableName)} AS tbl, ${sql.param(pred.col.name.name)} AS col, ${pred.col}::text AS value FROM ${spec.alias}`, + isLiteralParam(pred.to) + ? sql`SELECT ${sql.param(spec.tableName)} AS tbl, ${sql.param(pred.col.name.name)} AS col, ${asText(pred.to)} AS value` + : sql`SELECT ${sql.param(spec.tableName)} AS tbl, ${sql.param(pred.col.name.name)} AS col, ${asText(pred.col)} AS value FROM ${spec.alias}`, ); }); + // Decoy aliases claim each referenced table's bare name in the scope, so + // a real CTE alias whose ts-alias equals its table name suffixes to e.g. + // "users_2". Unlike pg, sqlite resolves a CTE's name inside its own body + // — `WITH users AS (SELECT … FROM users)` is a "circular reference" + // error there, so no CTE may shadow a table it selects from. The decoys + // are never emitted; they only occupy names. + const decoys = [...new Set(specs.map((s) => s.tableName))].map((n) => new Alias(n)); + return sql.withScope( - specs.map((s) => s.alias), + [...decoys, ...specs.map((s) => s.alias)], sql`WITH ${sql.join(ctes)} ${sql.join(unionParts, sql` UNION ALL `)}`, ); }; @@ -317,46 +336,26 @@ export const materializePredicateSet = ( return groups; }; -export type LiveIterationResult = { - rows: RowTypeToTsType[]; - // pg_current_snapshot() captured inside the txn — raw text. db.live() - // parses to Cursor; tests can inspect either form. - cursor: string; - // Closed set of watched values per (table, col) — propagated across - // equality classes. Ready to feed bus.subscribe() directly. - predicateSet: PredicateSet; -}; - -// One iteration of the live loop: open a REPEATABLE READ txn, snapshot the -// cursor, run the extractor, run the user query, commit. Caller handles the -// outer "yield + wait for matching event + repeat" loop. -export const runLiveIteration = async >( - db: Connection, - query: Q, -): Promise< - Q extends QueryBuilder - ? LiveIterationResult - : never -> => { - return db.transaction({ isolation: "repeatable read" }, async (tx) => { - const cursorResult = await tx.execute( - sql`SELECT pg_current_snapshot()::text AS cursor`, - ); - const cursor = (cursorResult.rows as { cursor: string }[])[0]!.cursor; - - // Important: traverse against a *single* finalize. Each finalize() mints - // fresh aliases, so the extractor SQL and the user query intentionally - // don't share alias identity — they're independently compiled. The - // extractor's aliases are sealed inside buildExtractor's withScope. - const finalized = query.finalize(); - const traversal = traverse(finalized); - const extractorSql = buildExtractor(sortAliases(traversal)); - const extractedResult = await tx.execute(extractorSql); - const extracted = extractedResult.rows as ExtractedRow[]; - const predicateSet = materializePredicateSet(extracted, traversal); - - const rows = await tx.execute(query); - - return { rows, cursor, predicateSet }; - }) as any; +// The dialect-neutral core of one live iteration: run the extractor, run +// the user query, return rows + watched predicate set. Each dialect's +// Executor.runLiveIteration wraps this with its consistency story (pg: +// REPEATABLE READ txn + snapshot cursor; sqlite: pre-read seq cursor). +export const runExtraction = async ( + conn: Connection, + query: QueryBuilder, +): Promise<{ rows: unknown; predicateSet: PredicateSet }> => { + // Important: traverse against a *single* finalize. Each finalize() mints + // fresh aliases, so the extractor SQL and the user query intentionally + // don't share alias identity — they're independently compiled. The + // extractor's aliases are sealed inside buildExtractor's withScope. + const finalized = query.finalize(); + const traversal = traverse(finalized); + const extractorSql = buildExtractor(sortAliases(traversal)); + const extractedResult = await conn.execute(extractorSql); + const extracted = extractedResult.rows as ExtractedRow[]; + const predicateSet = materializePredicateSet(extracted, traversal); + + const rows = await conn.execute(query); + + return { rows, predicateSet }; }; diff --git a/src/live/iteration.test.ts b/src/live/iteration.test.ts index 20dcec3f..7acf344d 100644 --- a/src/live/iteration.test.ts +++ b/src/live/iteration.test.ts @@ -2,11 +2,11 @@ import { test, expect } from "vitest"; import { Int8, Text } from "../types/postgres"; import { sql } from "../builder/sql"; import { setupDb, db, withinTransaction } from "../test-helpers"; -import { runLiveIteration } from "./extractor"; +import { PgExecutor } from "./pg/executor"; setupDb(); -test("runLiveIteration: returns rows + cursor + extracted rows from one txn", async () => { +test("PgExecutor.runLiveIteration: returns rows + cursor + extracted rows from one txn", async () => { // Outer txn must be at least as strong as runLiveIteration's request, or // the nested call rejects to avoid silently downgrading isolation. await withinTransaction(async (tx) => { @@ -41,15 +41,18 @@ test("runLiveIteration: returns rows + cursor + extracted rows from one txn", as .where(({ users }) => users.id.eq("1")) .select(({ users, dogs }) => ({ uid: users.id, dog: dogs.name })); - const { rows, cursor, predicateSet } = await runLiveIteration(tx, query); + // runLiveIteration goes entirely through the passed Connection; the + // executor's own exec fn is never touched, so a throwing dummy is fine. + const executor = new PgExecutor(db, () => Promise.reject(new Error("unused"))); + const { rows, cursor, predicateSet } = await executor.runLiveIteration(tx, query); // Original query: only Rex/Fido (user_id = 1) expect(rows.map((r) => r.dog).sort()).toEqual(["Fido", "Rex"]); expect(rows.every((r) => r.uid === "1")).toBe(true); - // Cursor is a non-empty pg_current_snapshot()::text - expect(typeof cursor).toBe("string"); - expect(cursor.length).toBeGreaterThan(0); + // Cursor is a parsed pg_current_snapshot() + expect(typeof cursor.xmin).toBe("bigint"); + expect(cursor.xmax >= cursor.xmin).toBe(true); // PredicateSet: users.id=1 from anchor, dogs.user_id=1 propagated. expect(predicateSet.get("users")?.get("id")).toEqual(new Set(["1"])); diff --git a/src/live/db-live.test.ts b/src/live/pg/db-live.test.ts similarity index 70% rename from src/live/db-live.test.ts rename to src/live/pg/db-live.test.ts index 34ba5815..45f0d70c 100644 --- a/src/live/db-live.test.ts +++ b/src/live/pg/db-live.test.ts @@ -1,8 +1,7 @@ import { test, expect, afterEach } from "vitest"; -import { Int8, Text } from "../types/postgres"; -import { sql } from "../builder/sql"; -import { conn, db, setupDb } from "../test-helpers"; -import { TypegresLiveEvents } from "./events"; +import { Int8, Text } from "../../types/postgres"; +import { sql } from "../../builder/sql"; +import { conn, db, setupDb } from "../../test-helpers"; import { setupLiveEvents } from "./test-helpers"; setupDb(); setupLiveEvents(); @@ -11,9 +10,6 @@ setupLiveEvents(); // visible to the bus's polling connection. afterEach tears down the // per-test artifacts. afterEach(async () => { - // db.stopLive may have already been called by the test — guarded by - // its idempotent semantics. - await conn.stopLive(); await conn.execute(sql`DROP TABLE IF EXISTS notes`); await conn.execute(sql`DROP TABLE IF EXISTS users`); await conn.execute(sql`DROP TABLE IF EXISTS dogs`); @@ -25,7 +21,7 @@ const makeNotesTable = async () => { user_id int8 NOT NULL, body text NOT NULL )`); - return class Notes extends db.Table("notes", { transformer: TypegresLiveEvents.makeTransformer() }) { + return class Notes extends db.Table("notes", { live: true }) { id = Int8.column({ nonNull: true, generated: true }); user_id = Int8.column({ nonNull: true }); body = Text.column({ nonNull: true }); @@ -40,13 +36,11 @@ const takeNext = async (iter: AsyncIterator): Promise => { test("yields current rows then re-yields on a matching commit (insert path)", async () => { const Notes = await makeNotesTable(); - await conn.startLive({ intervalMs: 25 }); - const iter = conn.live( - Notes.from() - .where(({ notes }) => notes.user_id.eq("1")) - .select(({ notes }) => ({ id: notes.id, body: notes.body })), - )[Symbol.asyncIterator](); + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq("1")) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); expect(await takeNext(iter)).toEqual([]); @@ -73,13 +67,11 @@ test("yields current rows then re-yields on a matching commit (insert path)", as test("UPDATE on a matching row re-yields with new values", async () => { const Notes = await makeNotesTable(); await Notes.insert({ user_id: "1", body: "first" }).execute(conn); - await conn.startLive({ intervalMs: 25 }); - const iter = conn.live( - Notes.from() - .where(({ notes }) => notes.user_id.eq("1")) - .select(({ notes }) => ({ id: notes.id, body: notes.body })), - )[Symbol.asyncIterator](); + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq("1")) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); const first = await takeNext(iter); expect(first[0]!.body).toBe("first"); @@ -101,13 +93,11 @@ test("DELETE on a matching row re-yields with the row removed", async () => { { user_id: "1", body: "keep-1" }, { user_id: "1", body: "delete-me" }, ).execute(conn); - await conn.startLive({ intervalMs: 25 }); - const iter = conn.live( - Notes.from() - .where(({ notes }) => notes.user_id.eq("1")) - .select(({ notes }) => ({ id: notes.id, body: notes.body })), - )[Symbol.asyncIterator](); + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq("1")) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); const first = await takeNext(iter); expect(first).toHaveLength(2); @@ -131,19 +121,17 @@ test("join: mutation on either side triggers re-yield (literal propagates across await conn.execute(sql`INSERT INTO users (id, name) OVERRIDING SYSTEM VALUE VALUES (5, 'alice')`); const Notes = await makeNotesTable(); - class Users extends db.Table("users", { transformer: TypegresLiveEvents.makeTransformer() }) { + class Users extends db.Table("users", { live: true }) { id = Int8.column({ nonNull: true, generated: true }); name = Text.column({ nonNull: true }); } - await conn.startLive({ intervalMs: 25 }); - const iter = conn.live( - Users.from() - .join(Notes, ({ users, notes }) => notes.user_id.eq(users.id)) - .where(({ users }) => users.id.eq("5")) - .select(({ users, notes }) => ({ user_name: users.name, body: notes.body })), - )[Symbol.asyncIterator](); + const iter = Users.from() + .join(Notes, ({ users, notes }) => notes.user_id.eq(users.id)) + .where(({ users }) => users.id.eq("5")) + .select(({ users, notes }) => ({ user_name: users.name, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); const first = await takeNext(iter); expect(first).toEqual([]); @@ -171,8 +159,3 @@ test("join: mutation on either side triggers re-yield (literal propagates across await iter.return?.(); }, 10_000); -test("throws if startLive wasn't called", async () => { - const Notes = await makeNotesTable(); - const iter = conn.live(Notes.from())[Symbol.asyncIterator](); - await expect(iter.next()).rejects.toThrow(/startLive/); -}); diff --git a/src/live/events-ddl.ts b/src/live/pg/events-ddl.ts similarity index 69% rename from src/live/events-ddl.ts rename to src/live/pg/events-ddl.ts index f50b9a4f..420e6e65 100644 --- a/src/live/events-ddl.ts +++ b/src/live/pg/events-ddl.ts @@ -4,8 +4,8 @@ // insert → database) — that cycle bit us at the chunk boundary // when vite bundled the site (TDZ during dynamic import). -import { sql, type Sql } from "../builder/sql"; -import type { Database } from "../database"; +import { sql, type Sql } from "../../builder/sql"; +import type { Connection, Database } from "../../database"; export const EVENTS_TABLE_NAME = "_typegres_live_events"; @@ -31,3 +31,14 @@ export const eventsTableSqlStatements = (database: Database): Sql[] => { sql`CREATE INDEX IF NOT EXISTS ${database.scopedIdent(`${EVENTS_TABLE_NAME}_xid_idx`)} ON ${database.scopedIdent(EVENTS_TABLE_NAME)} (xid)`, ]; }; + +// One-time pg schema setup for live queries (idempotent). Sqlite needs no +// setup — capture rides on RETURNING and events stay in memory. +export const ensurePgLiveEventsTable = async (conn: Connection): Promise => { + if (conn.database.dialect !== "postgres") { + throw new Error("ensurePgLiveEventsTable is pg-only — the sqlite live backend has no schema to install"); + } + for (const stmt of eventsTableSqlStatements(conn.database)) { + await conn.execute(stmt); + } +}; diff --git a/src/live/events.test.ts b/src/live/pg/events.test.ts similarity index 92% rename from src/live/events.test.ts rename to src/live/pg/events.test.ts index c694f079..7f393fe5 100644 --- a/src/live/events.test.ts +++ b/src/live/pg/events.test.ts @@ -1,14 +1,13 @@ import { test, expect, beforeAll, afterEach } from "vitest"; -import { Int8, Text } from "../types/postgres"; -import { sql } from "../builder/sql"; -import { conn, db, setupDb } from "../test-helpers"; -import { TypegresLiveEvents } from "./events"; +import { Int8, Text } from "../../types/postgres"; +import { sql } from "../../builder/sql"; +import { conn, db, setupDb } from "../../test-helpers"; import { setupLiveEvents } from "./test-helpers"; setupDb(); setupLiveEvents(); -class Foos extends db.Table("foos", { transformer: TypegresLiveEvents.makeTransformer() }) { +class Foos extends db.Table("foos", { live: true }) { id = Int8.column({ nonNull: true, generated: true }); name = Text.column({ nonNull: true }); qty = Int8.column(); @@ -59,7 +58,7 @@ test("insert RETURNING surfaces user columns through the wrap", async () => { }); test("delete emits one event per deleted row with before-image", async () => { - // Seed via raw SQL so the transformer doesn't fire for the setup rows. + // Seed via raw SQL so live capture doesn't fire for the setup rows. await conn.execute(sql`INSERT INTO foos (id, name) OVERRIDING SYSTEM VALUE VALUES (10, 'a'), (11, 'b'), (12, 'c')`); await Foos.delete().where(({ foos }) => foos.id.lt("12")).execute(conn); diff --git a/src/live/events.ts b/src/live/pg/events.ts similarity index 86% rename from src/live/events.ts rename to src/live/pg/events.ts index 0667d0e7..bc8207c3 100644 --- a/src/live/events.ts +++ b/src/live/pg/events.ts @@ -1,17 +1,17 @@ -import { Any, Jsonb } from "../types/postgres"; -import type { QueryTransformer } from "../table"; -import { sql, Ident, type Sql } from "../builder/sql"; -import type { Database } from "../database"; -import type { InsertBuilder } from "../builder/insert"; -import type { DeleteBuilder } from "../builder/delete"; -import type { UpdateBuilder} from "../builder/update"; -import { compileSetClauses } from "../builder/update"; -import { compileSelectList, type RowType } from "../builder/query"; +import { Any, Jsonb } from "../../types/postgres"; +import { sql, Ident, type Sql } from "../../builder/sql"; +import type { Database } from "../../database"; +import type { InsertBuilder } from "../../builder/insert"; +import type { DeleteBuilder } from "../../builder/delete"; +import type { UpdateBuilder} from "../../builder/update"; +import { compileSetClauses } from "../../builder/update"; +import { compileSelectList, type RowType } from "../../builder/query"; import { EVENTS_TABLE_NAME, eventsTableSqlStatements } from "./events-ddl"; -// Shadow table that captures every mutation against a "live"-enabled table. -// `makeTransformer` returns the per-op hook other tables opt in to: -// class Foos extends db.Table("foos", { transformer: TypegresLiveEvents.makeTransformer() }) +// Shadow table that captures every mutation against a `{ live: true }` +// table. PgExecutor applies wrapInsertOrDelete/wrapUpdate to those +// mutations at execute time — unconditionally, not bus-gated, because +// this table is the transport for other connections' pollers. export class TypegresLiveEvents { static readonly tableName = EVENTS_TABLE_NAME; @@ -21,14 +21,6 @@ export class TypegresLiveEvents { static createTableSqlStatements(database: Database): Sql[] { return eventsTableSqlStatements(database); } - - static makeTransformer(): QueryTransformer { - return { - insert: (s: InsertBuilder) => wrapInsertOrDelete(s, "after"), - delete: (s: DeleteBuilder) => wrapInsertOrDelete(s, "before"), - update: (s: UpdateBuilder) => wrapUpdate(s), - }; - } } // `jsonb_build_object('col', , ...)` over every Any-typed field @@ -87,12 +79,12 @@ const eventsInsertCte = (tableName: string, before: Sql, after: Sql, database: D // __typegres_events AS (INSERT INTO _typegres_live_events (xid, "table", before, after) // SELECT pg_current_xact_id(), ''::text, , FROM __typegres_cte) // SELECT FROM __typegres_cte -const wrapInsertOrDelete = ( +export const wrapInsertOrDelete = ( builder: InsertBuilder | DeleteBuilder, side: "before" | "after", ): Sql => { - const tableName = builder.tableName; - const database = builder.database; + const tableName = builder.table.tableName; + const database = builder.table.database; const liveKey = side === "before" ? T_LIVE_BEFORE : T_LIVE_AFTER; // returningMerge() has the same shape on both InsertBuilder and @@ -128,9 +120,9 @@ const wrapInsertOrDelete = ( // against just those rows. RETURNING in an UPDATE…FROM can reference both // the post-update target (``) and the FROM-clause CTE — that's how // we capture both images in one statement. -const wrapUpdate = (builder: UpdateBuilder): Sql => { - const tableName = builder.tableName; - const database = builder.database; +export const wrapUpdate = (builder: UpdateBuilder): Sql => { + const tableName = builder.table.tableName; + const database = builder.table.database; // Add both __typegres_live_before and __typegres_live_after to RETURNING. // After uses the post-update namespace refs (same as INSERT/DELETE); the diff --git a/src/live/pg/executor.ts b/src/live/pg/executor.ts new file mode 100644 index 00000000..a65c6799 --- /dev/null +++ b/src/live/pg/executor.ts @@ -0,0 +1,64 @@ +import { type CompiledSql, sql, type Sql } from "../../builder/sql"; +import type { Connection, Database } from "../../database"; +import type { AnyExecuteFn, QueryResult } from "../../drivers/types"; +import type { QueryBuilder, RowType } from "../../builder/query"; +import type { InsertBuilder } from "../../builder/insert"; +import type { UpdateBuilder } from "../../builder/update"; +import { + Executor, + isLiveMutation, + type LiveIterationResult, + type MutationBuilder, + type MutationOp, +} from "../../executor"; +import { runExtraction } from "../extractor"; +import { parseSnapshot } from "../snapshot"; +import { wrapInsertOrDelete, wrapUpdate } from "./events"; + +// Postgres execution context: mutations on `{ live: true }` tables are +// wrapped in the events-CTE chain (./events.ts) so they log to the shadow +// events table in the same statement — unconditionally, since that table +// transports to pollers on other connections. One class serves pooled +// and tx contexts (pg's MVCC already gives events commit-time visibility). +export class PgExecutor extends Executor { + constructor( + database: Database, + private execFn: AnyExecuteFn, + bound = false, + ) { + super(database, bound); + } + protected exec(compiled: CompiledSql): Promise { + return Promise.resolve(this.execFn(compiled)); + } + protected override runMutation(op: MutationOp, builder: MutationBuilder): Promise { + if (!isLiveMutation(builder)) { + return this.runStatement(builder); + } + const wrapped: Sql = + op === "update" + ? wrapUpdate(builder as UpdateBuilder) + : wrapInsertOrDelete(builder as InsertBuilder, op === "insert" ? "after" : "before"); + return this.runStatement(wrapped); + } + // Open a REPEATABLE READ txn, snapshot the cursor INSIDE it, run the + // extractor + user query — one consistent picture. Goes through + // conn.transaction() (not this executor's exec fn) for the bound context. + override runLiveIteration>( + conn: Connection, + query: Q, + ): Promise< + Q extends QueryBuilder + ? LiveIterationResult + : never + > { + return conn.transaction({ isolation: "repeatable read" }, async (tx) => { + const cursorResult = await tx.execute( + sql`SELECT pg_current_snapshot()::text AS cursor`, + ); + const cursor = parseSnapshot((cursorResult.rows as { cursor: string }[])[0]!.cursor); + const { rows, predicateSet } = await runExtraction(tx, query); + return { rows, cursor, predicateSet }; + }) as any; + } +} diff --git a/src/live/test-helpers.ts b/src/live/pg/test-helpers.ts similarity index 86% rename from src/live/test-helpers.ts rename to src/live/pg/test-helpers.ts index efd5481a..58ae3ea0 100644 --- a/src/live/test-helpers.ts +++ b/src/live/pg/test-helpers.ts @@ -1,6 +1,6 @@ import { afterEach, beforeAll } from "vitest"; -import { sql } from "../builder/sql"; -import { conn } from "../test-helpers"; +import { sql } from "../../builder/sql"; +import { conn } from "../../test-helpers"; import { TypegresLiveEvents } from "./events"; // Opt-in for live tests: creates `_typegres_live_events` once at the start diff --git a/src/live/snapshot.ts b/src/live/snapshot.ts index 106ab178..0b181e37 100644 --- a/src/live/snapshot.ts +++ b/src/live/snapshot.ts @@ -34,3 +34,12 @@ export const visible = (cursor: Cursor, xid: bigint): boolean => { if (xid >= cursor.xmax) { return false; } // not yet assigned/committed return !cursor.xip.has(xid); // in [xmin, xmax): visible iff not in-flight }; + +// A seq counter embedded in the snapshot shape (sqlite live): +// `visible(cursor, xid)` ⇔ `xid ≤ seq`. A slight fiction — no MVCC here — +// but it lets the bus's subscribe/backfill/visibility machinery run +// unmodified on integer event clocks. +export const seqCursor = (seq: bigint): Cursor => { + const next = seq + 1n; + return { xmin: next, xmax: next, xip: new Set() }; +}; diff --git a/src/live/sqlite/db-live.test.ts b/src/live/sqlite/db-live.test.ts new file mode 100644 index 00000000..7c9b4d83 --- /dev/null +++ b/src/live/sqlite/db-live.test.ts @@ -0,0 +1,429 @@ +import { test, expect, beforeAll, afterAll, afterEach } from "vitest"; +import { Database, type Connection } from "../../database"; +import { SqliteDriver } from "../../drivers/sqlite"; +import { DoSqliteDriver } from "../../drivers/do"; +import BetterSqlite3 from "better-sqlite3"; +import { sql } from "../../builder/sql"; +import { Integer, Real, Text } from "../../types/sqlite"; + +// End-to-end live queries over better-sqlite3 — mirrors the pg suite in +// ../pg/db-live.test.ts. No pollNow/interval anywhere: dispatch is +// synchronous with the mutation, so every re-yield below resolves without +// timers. The same backend drives DoSqliteDriver (Durable Objects). + +const db = new Database({ dialect: "sqlite" }); + +let driver: SqliteDriver; +let conn: Connection; + +beforeAll(async () => { + driver = await SqliteDriver.create(":memory:"); + conn = db.attach(driver); +}); + +afterAll(async () => { + await conn.close(); +}); + +afterEach(async () => { + await conn.execute(sql`DROP TABLE IF EXISTS notes`); + await conn.execute(sql`DROP TABLE IF EXISTS users`); + await conn.execute(sql`DROP TABLE IF EXISTS readings`); +}); + +const makeNotesTable = async () => { + await conn.execute(sql`CREATE TABLE notes ( + id INTEGER PRIMARY KEY, + user_id INTEGER NOT NULL, + body TEXT NOT NULL + )`); + return class Notes extends db.Table("notes", { live: true }) { + id = Integer.column({ nonNull: true }); + user_id = Integer.column({ nonNull: true }); + body = Text.column({ nonNull: true }); + }; +}; + +const takeNext = async (iter: AsyncIterator): Promise => { + const r = await iter.next(); + if (r.done) { throw new Error("iterator exhausted"); } + return r.value; +}; + +test("yields current rows then re-yields on a matching commit (insert path)", async () => { + const Notes = await makeNotesTable(); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); + + expect(await takeNext(iter)).toEqual([]); + + await Notes.insert({ id: 1, user_id: 1, body: "hello" }).execute(conn); + + const second = await takeNext(iter); + expect(second).toHaveLength(1); + expect(second[0]!.body).toBe("hello"); + + // Non-matching insert (user 99) shouldn't trigger a re-yield… + const racing = iter.next(); + await Notes.insert({ id: 2, user_id: 99, body: "irrelevant" }).execute(conn); + // …but the next matching insert should win the race. + await Notes.insert({ id: 3, user_id: 1, body: "world" }).execute(conn); + + const third = await racing; + expect(third.done).toBe(false); + const bodies = (third.value as { body: string }[]).map((r) => r.body).sort(); + expect(bodies).toEqual(["hello", "world"]); + + await iter.return?.(); +}, 10_000); + +test("UPDATE on a matching row re-yields with new values", async () => { + const Notes = await makeNotesTable(); + await Notes.insert({ id: 1, user_id: 1, body: "first" }).execute(conn); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); + + const first = await takeNext(iter); + expect(first[0]!.body).toBe("first"); + + await Notes.update() + .where(({ notes }) => notes.user_id.eq(1)) + .set(() => ({ body: "edited" })) + .execute(conn); + + const second = await takeNext(iter); + expect(second[0]!.body).toBe("edited"); + + await iter.return?.(); +}, 10_000); + +test("UPDATE moving a row OUT of the watched set re-yields (before-image path)", async () => { + const Notes = await makeNotesTable(); + await Notes.insert({ id: 1, user_id: 1, body: "mine" }).execute(conn); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); + + expect(await takeNext(iter)).toHaveLength(1); + + // After this update the row no longer matches user_id = 1. The + // after-image (user_id = 2) doesn't intersect the watched set — only + // the pre-SELECT'd before-image can trigger the re-yield. + await Notes.update() + .where(({ notes }) => notes.id.eq(1)) + .set(() => ({ user_id: 2 })) + .execute(conn); + + expect(await takeNext(iter)).toEqual([]); + + await iter.return?.(); +}, 10_000); + +test("DELETE on a matching row re-yields with the row removed", async () => { + const Notes = await makeNotesTable(); + await Notes.insert( + { id: 1, user_id: 1, body: "keep-1" }, + { id: 2, user_id: 1, body: "delete-me" }, + ).execute(conn); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); + + const first = await takeNext(iter); + expect(first).toHaveLength(2); + + await Notes.delete() + .where(({ notes }) => notes.body.eq("delete-me")) + .execute(conn); + + const second = await takeNext(iter); + expect(second).toHaveLength(1); + expect(second[0]!.body).toBe("keep-1"); + + await iter.return?.(); +}, 10_000); + +test("join: mutation on either side triggers re-yield (literal propagates across edge)", async () => { + await conn.execute(sql`CREATE TABLE users ( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL + )`); + const Notes = await makeNotesTable(); + + class Users extends db.Table("users", { live: true }) { + id = Integer.column({ nonNull: true }); + name = Text.column({ nonNull: true }); + } + + await Users.insert({ id: 5, name: "alice" }).execute(conn); + + const iter = Users.from() + .join(Notes, ({ users, notes }) => notes.user_id.eq(users.id)) + .where(({ users }) => users.id.eq(5)) + .select(({ users, notes }) => ({ user_name: users.name, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); + + const first = await takeNext(iter); + expect(first).toEqual([]); + + // Insert a note matching user 5 — should fire even though the predicate + // graph never had a literal anchor on notes.user_id directly. + await Notes.insert({ id: 1, user_id: 5, body: "hello" }).execute(conn); + + const second = await takeNext(iter); + expect(second).toHaveLength(1); + expect(second[0]!).toEqual({ + user_name: "alice", + body: "hello", + }); + + // Update on the users side also fires. + await Users.update() + .where(({ users }) => users.id.eq(5)) + .set(() => ({ name: "ALICE" })) + .execute(conn); + + const third = await takeNext(iter); + expect(third[0]!.user_name).toBe("ALICE"); + + await iter.return?.(); +}, 10_000); + +test("canonicalization parity: real + text anchors match captured images", async () => { + // If the extractor's CAST(col AS text) rendering ever diverged from the + // capture's json_object(CAST(col AS text)) rendering, the dispatcher's + // string matching would silently miss and these next() calls would hang + // (failing the test timeout). + await conn.execute(sql`CREATE TABLE readings ( + id INTEGER PRIMARY KEY, + sensor TEXT NOT NULL, + value REAL NOT NULL + )`); + class Readings extends db.Table("readings", { live: true }) { + id = Integer.column({ nonNull: true }); + sensor = Text.column({ nonNull: true }); + value = Real.column({ nonNull: true }); + } + + const bySensor = Readings.from() + .where(({ readings }) => readings.sensor.eq("temp")) + .select(({ readings }) => ({ value: readings.value })) + .live(conn)[Symbol.asyncIterator](); + const byValue = Readings.from() + .where(({ readings }) => readings.value.eq(1.5)) + .select(({ readings }) => ({ sensor: readings.sensor })) + .live(conn)[Symbol.asyncIterator](); + + expect(await takeNext(bySensor)).toEqual([]); + expect(await takeNext(byValue)).toEqual([]); + + await Readings.insert({ id: 1, sensor: "temp", value: 1.5 }).execute(conn); + + expect(await takeNext(bySensor)).toEqual([{ value: 1.5 }]); + expect(await takeNext(byValue)).toEqual([{ sensor: "temp" }]); + + await bySensor.return?.(); + await byValue.return?.(); +}, 10_000); + +test("mutations RETURNING still yields only user columns (images stripped)", async () => { + const Notes = await makeNotesTable(); + + const returned = await Notes.insert({ id: 1, user_id: 1, body: "hi" }) + .returning(({ notes }) => ({ id: notes.id, body: notes.body })) + .execute(conn); + expect(returned).toEqual([{ id: 1, body: "hi" }]); +}, 10_000); + + +test("cancelLiveSubscriptions ends an iterator that is MID-RERUN (not parked)", async () => { + const Notes = await makeNotesTable(); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id })) + .live(conn)[Symbol.asyncIterator](); + await takeNext(iter); + + const racing = iter.next(); + // The insert signals the parked sub synchronously (same-tick flush), + // queueing the generator's rerun; cancelling right after lands while + // the rerun is in flight — there is no parked Subscription to reject, + // only the epoch check can end the iterator. + await Notes.insert({ id: 1, user_id: 1, body: "wake" }).execute(conn); + conn.cancelLiveSubscriptions(); + expect((await racing).done).toBe(true); +}, 10_000); + +test("cancelLiveSubscriptions releases a parked consumer cleanly", async () => { + const Notes = await makeNotesTable(); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id })) + .live(conn)[Symbol.asyncIterator](); + await takeNext(iter); + + const parked = iter.next(); + conn.cancelLiveSubscriptions(); + expect((await parked).done).toBe(true); +}, 10_000); + +test("close() stops the live engine and releases a parked consumer", async () => { + const ownDriver = await SqliteDriver.create(":memory:"); + const ownConn = db.attach(ownDriver); + await ownConn.execute(sql`CREATE TABLE notes ( + id INTEGER PRIMARY KEY, + user_id INTEGER NOT NULL, + body TEXT NOT NULL + )`); + class Notes extends db.Table("notes", { live: true }) { + id = Integer.column({ nonNull: true }); + user_id = Integer.column({ nonNull: true }); + body = Text.column({ nonNull: true }); + } + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id })) + .live(ownConn)[Symbol.asyncIterator](); + await takeNext(iter); + + const parked = iter.next(); + await ownConn.close(); + expect((await parked).done).toBe(true); +}, 10_000); + +test("live over DoSqliteDriver (fake SqlStorage backed by better-sqlite3)", async () => { + // Faithful-enough DurableObjectStorage fake: synchronous exec with + // positional bindings, native JS values out, SqlStorage's BigInt + // rejection, and the storage.transaction() protocol (workerd rejects + // SQL BEGIN — see DoStorageLike). + const raw = new BetterSqlite3(":memory:"); + const storage = { + sql: { + exec(query: string, ...bindings: unknown[]) { + if (bindings.some((b) => typeof b === "bigint")) { + throw new TypeError("SqlStorage does not support BigInt bindings"); + } + if (/^\s*(BEGIN|COMMIT|ROLLBACK|SAVEPOINT)/i.test(query)) { + throw new Error("To execute a transaction, please use the state.storage.transaction() API"); + } + const stmt = raw.prepare(query); + const rows = stmt.reader ? (stmt.all(...bindings) as { [k: string]: unknown }[]) : (stmt.run(...bindings), []); + return { toArray: () => rows }; + }, + }, + async transaction(cb: () => Promise): Promise { + raw.exec("BEGIN"); + try { + const result = await cb(); + raw.exec("COMMIT"); + return result; + } catch (e) { + raw.exec("ROLLBACK"); + throw e; + } + }, + }; + const doConn = db.attach(new DoSqliteDriver(storage)); + + await doConn.execute(sql`CREATE TABLE notes ( + id INTEGER PRIMARY KEY, + user_id INTEGER NOT NULL, + body TEXT NOT NULL + )`); + class Notes extends db.Table("notes", { live: true }) { + id = Integer.column({ nonNull: true }); + user_id = Integer.column({ nonNull: true }); + body = Text.column({ nonNull: true }); + } + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(doConn)[Symbol.asyncIterator](); + + expect(await takeNext(iter)).toEqual([]); + await Notes.insert({ id: 1, user_id: 1, body: "from-do" }).execute(doConn); + expect(await takeNext(iter)).toEqual([{ id: 1, body: "from-do" }]); + + await iter.return?.(); + raw.close(); +}, 10_000); + + +test("transaction: captured events are withheld mid-tx and flush at COMMIT", async () => { + const Notes = await makeNotesTable(); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); + expect(await takeNext(iter)).toEqual([]); + + const racing = iter.next(); + await conn.transaction(async (tx) => { + await Notes.insert({ id: 1, user_id: 1, body: "in-tx" }).execute(tx); + await Notes.insert({ id: 2, user_id: 1, body: "also-in-tx" }).execute(tx); + // Events buffer until COMMIT — the subscriber must not have been + // signaled yet. (The signal would have resolved `racing` well within + // one microtask; racing an immediately-resolving probe is enough.) + const state = await Promise.race([ + racing.then(() => "resolved"), + Promise.resolve().then(() => "pending"), + ]); + expect(state).toBe("pending"); + }); + + const woke = await racing; + expect(woke.done).toBe(false); + expect((woke.value as { body: string }[]).map((r) => r.body).sort()).toEqual(["also-in-tx", "in-tx"]); + + await iter.return?.(); +}, 10_000); + +test("transaction: ROLLBACK discards captured events", async () => { + const Notes = await makeNotesTable(); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); + expect(await takeNext(iter)).toEqual([]); + + const racing = iter.next(); + await conn + .transaction(async (tx) => { + await Notes.insert({ id: 1, user_id: 1, body: "doomed" }).execute(tx); + throw new Error("__rollback__"); + }) + .catch((e) => { + if ((e as Error).message !== "__rollback__") { throw e; } + }); + + // The rolled-back insert must not have signaled; the next commit wins + // the race and its rowset must not contain the doomed row. + await Notes.insert({ id: 2, user_id: 1, body: "committed" }).execute(conn); + const woke = await racing; + expect(woke.done).toBe(false); + expect(woke.value).toEqual([{ id: 2, body: "committed" }]); + + await iter.return?.(); +}, 10_000); + +test("transaction({ isolation }) is rejected on sqlite", async () => { + await expect( + conn.transaction({ isolation: "serializable" }, async () => {}), + ).rejects.toThrow(/pg-only/); +}); diff --git a/src/live/sqlite/do-test-modules.d.ts b/src/live/sqlite/do-test-modules.d.ts new file mode 100644 index 00000000..94ae2e0c --- /dev/null +++ b/src/live/sqlite/do-test-modules.d.ts @@ -0,0 +1,26 @@ +// Minimal duck-typed declarations for the workerd virtual modules the +// *.do-test.ts suite imports — same philosophy as DoSqliteDriver's +// SqlStorageLike: declare only the surface we use, so these files +// typecheck under the MAIN tsconfig (node ambient globals) instead of +// needing a second TS project with @cloudflare/workers-types (whose +// globals conflict with @types/node). The real implementations are +// provided at runtime by @cloudflare/vitest-pool-workers / workerd. + +declare module "cloudflare:test" { + import type { DoStorageLike } from "../../drivers/do"; + + export const env: { + TEST_DO: { getByName(name: string): DurableObjectStubLike }; + }; + export interface DurableObjectStubLike { + readonly __brand?: "DurableObjectStub"; + } + export function runInDurableObject( + stub: DurableObjectStubLike, + callback: (instance: unknown, state: { storage: DoStorageLike }) => T | Promise, + ): Promise; +} + +declare module "cloudflare:workers" { + export class DurableObject {} +} diff --git a/src/live/sqlite/do-test.worker.ts b/src/live/sqlite/do-test.worker.ts new file mode 100644 index 00000000..3ebae3d0 --- /dev/null +++ b/src/live/sqlite/do-test.worker.ts @@ -0,0 +1,13 @@ +// Minimal worker entry for the workerd vitest project: exports the +// SQLite-backed DO class bound as TEST_DO. The *.do-test.ts tests never +// route requests through fetch — they enter the DO via cloudflare:test's +// runInDurableObject and drive typegres against its storage directly. +import { DurableObject } from "cloudflare:workers"; + +export class TestDO extends DurableObject {} + +export default { + fetch(): Response { + return new Response("test worker"); + }, +}; diff --git a/src/live/sqlite/executor.ts b/src/live/sqlite/executor.ts new file mode 100644 index 00000000..79e51404 --- /dev/null +++ b/src/live/sqlite/executor.ts @@ -0,0 +1,169 @@ +import { compile, type CompiledSql, Ident, sql, type Sql } from "../../builder/sql"; +import { canonicalText } from "../canonical"; +import type { QueryBuilder, RowType } from "../../builder/query"; +import type { Connection, Database } from "../../database"; +import type { QueryResult, SyncDriver } from "../../drivers/types"; +import type { InsertBuilder } from "../../builder/insert"; +import type { UpdateBuilder } from "../../builder/update"; +import { + Executor, + isLiveMutation, + type LiveIterationResult, + type MutationBuilder, + type MutationOp, +} from "../../executor"; +import { runExtraction } from "../extractor"; +import { seqCursor } from "../snapshot"; +import { type Bus, parseEventPairs } from "../bus"; +import { SqlValue } from "../../types/sql-value"; +import { Text } from "../../types/sqlite"; + +// Mutation-image capture for sqlite live. Mutations on `{ live: true }` +// tables get an image column merged into RETURNING at execute time — no +// shadow table (unlike pg, whose events table transports to other +// connections' pollers): +// +// insert → after-image (RETURNING sees the inserted row) +// delete → before-image (RETURNING evaluates against the deleted row) +// update → after-image; the before-image comes from a pre-SELECT +// (sqlite RETURNING can't see OLD values) +// +// Values render through canonicalText — must stay aligned with the +// extractor's rendering (see ../canonical.ts). + +// One column name serves before- and after-images alike: the matcher +// only consumes the union of (col, value) pairs. +const T_LIVE_IMAGE = "__typegres_live_image"; + +// A mutation's change events, unstamped — onCommit assigns xids from the +// driver's statement clock when it pushes them to the bus. +export type CapturedEvent = { table: string; pairs: [string, string][] }; + +// `json_object('col', , ...)` over every column field +// (SqlValue instances) of the row instance. +const buildImageJson = (tableInstance: { [c: string]: unknown }): Sql => { + const args = Object.entries(tableInstance) + .filter((e): e is [string, SqlValue] => e[1] instanceof SqlValue) + .flatMap(([c, v]) => [sql.param(c), canonicalText(v.toSql(), "sqlite")]); + return sql`json_object(${sql.join(args)})`; +}; + +// returningMerge callback adding the image column; runMutation strips it +// from result rows before they reach deserializeRows. +const imageReturning = + (tableName: string) => + (ns: object): RowType => { + const tableInstance = (ns as { [k: string]: object })[tableName] as { [c: string]: unknown }; + return { [T_LIVE_IMAGE]: Text.from(buildImageJson(tableInstance)) }; + }; + +// The UPDATE before-image: select the image of every row the UPDATE's +// WHERE matches, run synchronously just before the UPDATE — same state, +// no transaction needed (single writer, no awaits between). `.where(true)` +// (matchAll) leaves `where` undefined; emit WHERE TRUE explicitly. +// +// Why two statements instead of a before-CTE on the UPDATE: verified on +// sqlite 3.50, a CTE referenced only from RETURNING evaluates lazily +// DURING the update scan and reads post-update values (MATERIALIZED or +// not). The one sound single-statement form (MATERIALIZED + FROM +// rowid-join + subquery RETURNING) silently degrades to post-images if +// any piece is dropped, and breaks on WITHOUT ROWID tables. +const buildUpdatePreImageSelect = (builder: UpdateBuilder): Sql => { + const finalized = builder.finalize(); + const { alias, where, instance } = finalized.opts; + const database = builder.table.database; + const whereClause = where ? where.toSql() : sql`TRUE`; + return sql.withScope( + [alias], + sql`SELECT ${buildImageJson(instance as { [c: string]: unknown })} AS ${new Ident(T_LIVE_IMAGE)} FROM ${database.scopedIdent(builder.table.tableName)} AS ${alias} WHERE ${whereClause}`, + ); +}; + +// The sqlite live execution context (pooled and transaction-bound). +// Captured events buffer into #pending and flush at commit: +// - unbound: sqlite is in autocommit — each statement is its own +// transaction — so runMutation invokes onCommit itself, and events +// reach the bus in the same tick as the mutation. +// - bound: Connection.transaction() fires onCommit after COMMIT (or +// onRollback, discarding) — the explicit version of the commit-time +// event visibility pg gets from MVCC. Nested transaction() calls +// flatten onto the same executor, so the buffer spans to the +// outermost commit. +export class SqliteLiveExecutor extends Executor { + #pending: CapturedEvent[] = []; + + constructor( + database: Database, + private driver: SyncDriver, + private bus: Bus, + bound = false, + ) { + super(database, bound); + } + + protected exec(compiled: CompiledSql): Promise { + return Promise.resolve(this.driver.executeSync(compiled)); + } + + protected override runMutation(op: MutationOp, builder: MutationBuilder): Promise { + if (!isLiveMutation(builder)) { + return this.runStatement(builder); + } + const { executeSync } = this.driver; + const table = builder.table.tableName; + // returningMerge has the same shape on all three builders. + const merged = (builder as InsertBuilder).returningMerge(imageReturning(table)); + const compiled = compile(merged, { database: this.database }); + const events: CapturedEvent[] = []; + if (op === "update") { + const preImage = compile(buildUpdatePreImageSelect(builder as UpdateBuilder), { database: this.database }); + for (const row of executeSync(preImage).rows) { + events.push({ table, pairs: parseEventPairs(row[T_LIVE_IMAGE] ?? null, null) }); + } + } + const result = executeSync(compiled); + const rows = result.rows.map((row) => { + const { [T_LIVE_IMAGE]: image, ...rest } = row; + events.push({ table, pairs: parseEventPairs(image ?? null, null) }); + return rest; + }); + this.#pending.push(...events); + if (!this.bound) { + // Autocommit: this statement IS its transaction. + this.onCommit(); + } + // INVARIANT: pre-image read → mutation → flush is one synchronous + // frame. Keep this method await-free, or concurrent mutations could + // interleave between statements and in #pending. + return Promise.resolve({ rows }); + } + + // Stamp with the driver's statement clock — the tick of the statement + // making the events visible (the mutation, or COMMIT), so a mutation's + // (or whole transaction's) events share one xid, like pg's per-txn xids. + override onCommit(): void { + const pending = this.#pending; + this.#pending = []; + this.bus.ingest(pending.map((e) => ({ xid: this.driver.liveSeq, ...e }))); + } + + override onRollback(): void { + this.#pending = []; + } + + // No txn: readCursor runs BEFORE the queries, so any event stamped + // while they run is "not yet seen" and trips the subscribe-time + // backfill check. + override async runLiveIteration>( + conn: Connection, + query: Q, + ): Promise< + Q extends QueryBuilder + ? LiveIterationResult + : never + > { + const cursor = seqCursor(this.driver.liveSeq); + const { rows, predicateSet } = await runExtraction(conn, query); + return { rows, cursor, predicateSet } as any; + } +} diff --git a/src/live/sqlite/live.do-test.ts b/src/live/sqlite/live.do-test.ts new file mode 100644 index 00000000..8f22bc76 --- /dev/null +++ b/src/live/sqlite/live.do-test.ts @@ -0,0 +1,87 @@ +// Live queries against a REAL Durable Object's storage, inside workerd +// (the vitest "workerd" project; `npm run test:do` to run just these). +// The node suite's DoSqliteDriver test uses a better-sqlite3-backed +// fake; this is the genuine article. The cloudflare:* modules are +// duck-type declared in ./do-test-modules.d.ts. +import { test, expect } from "vitest"; +import { env, runInDurableObject } from "cloudflare:test"; +import { Database } from "../../database"; +import { DoSqliteDriver } from "../../drivers/do"; +import { sql } from "../../builder/sql"; +import { Integer, Text } from "../../types/sqlite"; + +const db = new Database({ dialect: "sqlite" }); + +class Notes extends db.Table("notes", { live: true }) { + id = Integer.column({ nonNull: true }); + user_id = Integer.column({ nonNull: true }); + body = Text.column({ nonNull: true }); +} + +const takeNext = async (iter: AsyncIterator): Promise => { + const r = await iter.next(); + if (r.done) { throw new Error("iterator exhausted"); } + return r.value; +}; + +test("live insert/update/delete round-trip on real DO SqlStorage", async () => { + const stub = env.TEST_DO.getByName("live-round-trip"); + await runInDurableObject(stub, async (_instance, state) => { + const conn = db.attach(new DoSqliteDriver(state.storage)); + await conn.execute(sql`CREATE TABLE notes ( + id INTEGER PRIMARY KEY, + user_id INTEGER NOT NULL, + body TEXT NOT NULL + )`); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(conn)[Symbol.asyncIterator](); + + expect(await takeNext(iter)).toEqual([]); + + await Notes.insert({ id: 1, user_id: 1, body: "from-real-do" }).execute(conn); + expect(await takeNext(iter)).toEqual([{ id: 1, body: "from-real-do" }]); + + await Notes.update() + .where(({ notes }) => notes.id.eq(1)) + .set(() => ({ body: "edited" })) + .execute(conn); + expect(await takeNext(iter)).toEqual([{ id: 1, body: "edited" }]); + + // Before-image path: the row leaves the watched set. + await Notes.update() + .where(({ notes }) => notes.id.eq(1)) + .set(() => ({ user_id: 2 })) + .execute(conn); + expect(await takeNext(iter)).toEqual([]); + + await iter.return?.(); + }); +}); + +test("transaction: buffered events flush at COMMIT on real DO", async () => { + const stub = env.TEST_DO.getByName("live-tx"); + await runInDurableObject(stub, async (_instance, state) => { + const conn = db.attach(new DoSqliteDriver(state.storage)); + await conn.execute(sql`CREATE TABLE notes ( + id INTEGER PRIMARY KEY, + user_id INTEGER NOT NULL, + body TEXT NOT NULL + )`); + + const iter = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ body: notes.body })) + .live(conn)[Symbol.asyncIterator](); + expect(await takeNext(iter)).toEqual([]); + + await conn.transaction(async (tx) => { + await Notes.insert({ id: 1, user_id: 1, body: "tx" }).execute(tx); + }); + expect(await takeNext(iter)).toEqual([{ body: "tx" }]); + + await iter.return?.(); + }); +}); diff --git a/src/table.ts b/src/table.ts index ca7cc200..ab84fe27 100644 --- a/src/table.ts +++ b/src/table.ts @@ -1,6 +1,6 @@ -import type { BoundSql, Ident, Sql } from "./builder/sql"; +import type { BoundSql, Ident } from "./builder/sql"; import type { Database } from "./database"; -import type { Fromable} from "./builder/query"; +import type { Fromable } from "./builder/query"; import { QueryBuilder } from "./builder/query"; import { DeleteBuilder } from "./builder/delete"; import { UpdateBuilder } from "./builder/update"; @@ -8,18 +8,11 @@ import { InsertBuilder } from "./builder/insert"; import { isColumn } from "./types/sql-value"; import type { InsertRow } from "./types/runtime"; -// A per-op SQL rewrite hook. Tables opt in via the `transformer` option; -// each mutation builder runs its own slot at bind() time, replacing its -// default-finalized SQL with whatever the hook returns. Used by the live -// system to wrap mutations in event-emitting CTE chains. -export type QueryTransformer = { - insert?: (builder: InsertBuilder) => Sql; - update?: (builder: UpdateBuilder) => Sql; - delete?: (builder: DeleteBuilder) => Sql; -}; - export type TableOptions = { - transformer?: QueryTransformer; + // Opt this table into live change capture: the dialect executor rewrites + // its mutations at execute time (sqlite: image-RETURNING + in-memory + // events; pg: the events-CTE chain logging to the shadow table). + live?: boolean; }; // Concrete tables extend `Table(name)` and declare columns as field @@ -45,9 +38,8 @@ export abstract class TableBase { // enforces same-db at bind time. Non-optional — tables are always // scoped to a Database. static readonly database: Database; - // Mutation builders on this table run this at bind() time and on each - // raw result row before deserialization. Default none. - static readonly transformer: QueryTransformer | undefined = undefined; + // Live change-capture opt-in (see TableOptions.live). + static readonly live: boolean = false; // Per-scope tag carried by `Table.scope(ctx)`. Each `scope()` call // mints an anonymous subclass that overrides this static with the // supplied value; hydrated row instances read it through their @@ -182,8 +174,7 @@ export const Table = ( [name]: class extends TableBase { static override readonly tableName = name; static override readonly tsAlias = name; - static override readonly transformer: QueryTransformer | undefined = - opts.transformer; + static override readonly live: boolean = opts.live ?? false; // Default value remains `undefined` until `scope()` overrides via // subclass; the type narrowing is purely compile-time. static override readonly context: C = undefined as C; diff --git a/src/test-helpers.ts b/src/test-helpers.ts index ff0232d2..e29abe5d 100644 --- a/src/test-helpers.ts +++ b/src/test-helpers.ts @@ -28,7 +28,9 @@ export const setupDb = (): void => { max: 1, options: `-csearch_path=${schema}`, }); - conn = db.attach(driver); + // Fast poll cadence for live suites; harmless elsewhere — the pg + // poller only starts on first .live() use. + conn = db.attach(driver, { intervalMs: 25 }); await conn.execute(sql`DROP SCHEMA IF EXISTS ${db.scopedIdent(schema)} CASCADE`); await conn.execute(sql`CREATE SCHEMA ${db.scopedIdent(schema)}`); }); diff --git a/vitest.config.ts b/vitest.config.ts index cbe6021f..fdeb4b40 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -1,13 +1,16 @@ -import { defineConfig } from "vitest/config"; +import { defineConfig, type ViteUserConfig } from "vitest/config"; import { fileURLToPath } from "node:url"; import swc from "unplugin-swc"; +import { cloudflareTest } from "@cloudflare/vitest-pool-workers"; // `demo.ts` uses bare package imports (`from "typegres"`) so the playground // snippet reads naturally. Alias those to the source so the demo test can // run against src without a build step. const src = fileURLToPath(new URL("./src", import.meta.url)); -export default defineConfig({ +// Shared by both projects (projects are separate Vite configs — root-level +// resolve/plugins don't inherit). +const shared = { resolve: { // Order matters — Vite matches aliases longest-first, so subpath // entries come before the bare `typegres` root. @@ -21,7 +24,7 @@ export default defineConfig({ // codegen'd methods). Vite 8's default Oxc transform leaves them as-is, // which trips the Node runtime with a SyntaxError. SWC lowers them. // `oxc: false` disables the default; SWC owns the TS pipeline. - oxc: false, + oxc: false as const, plugins: [ swc.vite({ jsc: { @@ -31,19 +34,55 @@ export default defineConfig({ }, }), ], +} satisfies ViteUserConfig; + +export default defineConfig({ test: { // Bumped from the 5s default: pg-backed tests hit their limit under // slow containers (act/docker on a single vCPU) even though they run // in ~50ms locally. testTimeout: 15000, - exclude: [ - "**/node_modules/**", - "**/dist/**", - "**/.claude/**", - "**/.direnv/**", - "**/examples/**", - "packages/**", - "site/**", + projects: [ + { + ...shared, + test: { + name: "node", + exclude: [ + // Workerd-project tests; the default include glob wouldn't + // match them anyway, but keep the partition explicit. + "**/*.do-test.ts", + "**/node_modules/**", + "**/dist/**", + "**/.claude/**", + "**/.direnv/**", + "**/examples/**", + "packages/**", + "site/**", + ], + }, + }, + // Real-Durable-Object tests (src/**/*.do-test.ts) run INSIDE workerd; + // the cloudflare:* modules they import are duck-type declared in + // src/live/sqlite/do-test-modules.d.ts. + { + ...shared, + plugins: [ + ...shared.plugins, + cloudflareTest({ + main: "./src/live/sqlite/do-test.worker.ts", + miniflare: { + compatibilityDate: "2025-01-01", + durableObjects: { + TEST_DO: { className: "TestDO", useSQLite: true }, + }, + }, + }), + ], + test: { + name: "workerd", + include: ["src/**/*.do-test.ts"], + }, + }, ], }, });