diff --git a/src/sqlite/client.ts b/src/sqlite/client.ts
deleted file mode 100644
index e69de29..0000000
diff --git a/src/sqlite/index.ts b/src/sqlite/index.ts
index 87abfbc..6f28df4 100644
--- a/src/sqlite/index.ts
+++ b/src/sqlite/index.ts
@@ -1,8 +1,7 @@
import sqlite3InitModule, {
Database,
type SAHPoolUtil,
- type Sqlite3Static,
- type WasmPointer
+ type Sqlite3Static
} from '@libsql/libsql-wasm-experimental'
import * as Comlink from 'comlink'
import { desc, gt } from 'drizzle-orm'
@@ -33,6 +32,7 @@ const subscribers = new Map<
const changeLogSubscribers = new Set
()
let logCursor = 0
const path = 'file:local.db'
+
const initClient = async () => {
try {
sqlite3 = await sqlite3InitModule({})
@@ -40,8 +40,8 @@ const initClient = async () => {
name: path,
initialCapacity: 10
})
+ await poolUtil.wipeFiles()
;[client, db] = createClient({ url: path, poolUtil }, sqlite3)
-
await client.execute(`PRAGMA foreign_keys = ON;`)
await runMigrations(client)
@@ -242,7 +242,7 @@ const opCounts: Record<'INSERT' | 'UPDATE' | 'DELETE', number> = {
}
let lastWindow = performance.now()
-const windowSize = 1000 // ms
+const windowSize = 1000
const notifyChangeLogSubscribers = (changes: schema.ChangeLog[]) => {
for (const { op_type } of changes) {
@@ -274,43 +274,13 @@ const notifyChangeLogSubscribers = (changes: schema.ChangeLog[]) => {
}
}
-const changeLogQueue: schema.ChangeLog[] = []
-let flushTimer: ReturnType | null = null
-const FLUSH_DELAY_MS = 10
-
-function flushChangeLogs() {
- const batch = changeLogQueue.splice(0, changeLogQueue.length)
- if (batch.length === 0) return
-
- notifyChangeLogSubscribers(batch)
- dispatchChangeNotificationBatched(batch)
-}
-
-const notifyBuffered = () => {
+const notifyBatched = () => {
const { rows } = getNewChangeLogsSync(logCursor)
if (rows.length === 0) return
log(
`Worker: Fetched ${rows.length} change logs since last cursor ${logCursor}`
)
- changeLogQueue.push(...rows)
- const lastLog = rows[rows.length]
- logCursor = rows[0].id
- if (!flushTimer) {
- flushTimer = setTimeout(() => {
- flushChangeLogs()
- flushTimer = null
- }, FLUSH_DELAY_MS)
- }
-}
-
-const notifyBatched = () => {
- const { rows } = getNewChangeLogsSync(logCursor)
- if (rows.length === 0) return
- // log(
- // `Worker: Fetched ${rows.length} change logs since last cursor ${logCursor}`
- // )
-
logCursor = rows[0].id
notifyChangeLogSubscribers(rows)
dispatchChangeNotificationBatched(rows)
@@ -328,81 +298,16 @@ const notify = () => {
}
}
-const setupChangeNotificationSystem = (sqlite3: Sqlite3Static) => {
- const changeLogBuffer: Array<{
- operation: 'INSERT' | 'UPDATE' | 'DELETE'
- dbName: string
- tableName: string
- rowid: string
- }> = []
-
- const OpNames: Record = {
- 9: 'DELETE',
- 18: 'INSERT',
- 23: 'UPDATE'
- }
-
- sqlite3.capi.sqlite3_update_hook(
- db.pointer! as WasmPointer,
- (
- _userCtx: number,
- op: 9 | 18 | 23,
- dbName: string,
- tableName: string,
- rowid: BigInt
- ) => {
- changeLogBuffer.push({
- operation: OpNames[op],
- dbName,
- tableName,
- rowid: rowid.toString()
- })
- },
- 0 as WasmPointer
- )
-
- sqlite3.capi.sqlite3_commit_hook(
- db.pointer! as WasmPointer,
- (_cbArg: WasmPointer) => {
- if (changeLogBuffer.length === 0) {
- log('No changes in transaction buffer')
- } else {
- for (const rec of changeLogBuffer) {
- log(
- `→ ${rec.operation} on ${rec.dbName}.${
- rec.tableName
- } (rowid=${rec.rowid.toString()})`
- )
- }
-
- const { rows } = getNewChangeLogsSync(logCursor)
- log(
- `Worker: Fetched ${rows.length} change logs since last cursor ${logCursor}`
- )
- if (rows.length === 0) return 0
- notifyChangeLogSubscribers(rows)
- logCursor = rows[0].id
- for (const rec of rows) {
- dispatchChangeNotification(rec)
- }
- }
-
- changeLogBuffer.length = 0
- return 0
- },
- 0 as WasmPointer
- )
-}
-
const onClientReady = async () => {
log('Worker: Client is ready!')
-
+ const all = await drizzleClient
+ .select({ id: schema.changeLog.id })
+ .from(schema.changeLog)
const [row] = await drizzleClient
.select({ id: schema.changeLog.id })
.from(schema.changeLog)
.orderBy(desc(schema.changeLog.id))
.limit(1)
-
logCursor = row ? row.id - 1000 : 0
const { rows } = await getNewChangeLogs(logCursor)
notifyChangeLogSubscribers(rows)
@@ -457,7 +362,6 @@ const getNewChangeLogsSync = (lastSeenLogId: number) => {
ORDER BY id DESC`,
[lastSeenLogId]
)
-
const rows = result.rows as any as schema.ChangeLog[]
return { rows }
@@ -578,18 +482,11 @@ function logFileSystemSupport() {
]
checks.forEach(({ name, supported }) => {
- console.log(`${name}: ${supported ? '✅ supported' : '❌ not supported'}`)
+ log(`${name}: ${supported ? '✅ supported' : '❌ not supported'}`)
})
}
-// Example usage:
-
logFileSystemSupport()
-// onconnect = function (event) {
-// console.log('connect!')
-// const port = event.ports[0]
-
-// }
const portApi = {
...api
diff --git a/src/sqlite/schema.ts b/src/sqlite/schema.ts
index 6c27ef1..532cfe3 100644
--- a/src/sqlite/schema.ts
+++ b/src/sqlite/schema.ts
@@ -1,4 +1,4 @@
-import { sql, relations } from 'drizzle-orm'
+import { relations, sql } from 'drizzle-orm'
import { integer, primaryKey, sqliteTable, text } from 'drizzle-orm/sqlite-core'
export const users = sqliteTable('users', {
@@ -10,10 +10,10 @@ export const users = sqliteTable('users', {
picture: text('picture'),
bio: text('bio'),
location: text('location'),
- createdAt: text('created_at')
+ created_at: text('created_at')
.notNull()
.default(sql`(CURRENT_TIMESTAMP)`),
- isActive: integer('is_active', { mode: 'boolean' }).default(true)
+ is_active: integer('is_active', { mode: 'boolean' }).default(true)
})
export type User = typeof users.$inferSelect
export type NewUser = typeof users.$inferInsert
@@ -22,31 +22,47 @@ export const posts = sqliteTable('posts', {
id: integer('id', { mode: 'number' })
.primaryKey({ autoIncrement: true })
.notNull(),
- authorId: integer('author_id', { mode: 'number' })
+ author_id: integer('author_id', { mode: 'number' })
.notNull()
.references(() => users.id, { onDelete: 'cascade' }),
title: text('title').notNull(),
body: text('body').notNull(),
- createdAt: text('created_at')
+ created_at: text('created_at')
.notNull()
.default(sql`(CURRENT_TIMESTAMP)`),
- updatedAt: text('updated_at')
+ updated_at: text('updated_at')
.notNull()
.default(sql`(CURRENT_TIMESTAMP)`)
})
export type Post = typeof posts.$inferSelect
export type NewPost = typeof posts.$inferInsert
-// export const usersRelations = relations(users, ({ many }) => ({
-// posts: many(posts)
-// }))
+export const usersRelations = relations(users, ({ many }) => ({
+ posts: many(posts)
+}))
-// export const postsRelations = relations(posts, ({ one }) => ({
-// author: one(users, {
-// fields: [posts.authorId],
-// references: [users.id]
-// })
-// }))
+export const postsRelations = relations(posts, ({ one }) => ({
+ author: one(users, {
+ fields: [posts.author_id],
+ references: [users.id]
+ })
+}))
+
+export const cells = sqliteTable(
+ 'cells',
+ {
+ x: integer('x', { mode: 'number' }).notNull(),
+ y: integer('y', { mode: 'number' }).notNull(),
+ alive: integer('alive', { mode: 'boolean' }).default(false).notNull()
+ },
+ table => [primaryKey({ columns: [table.x, table.y] })]
+)
+export type Cells = typeof cells.$inferSelect
+export type NewCells = typeof cells.$inferInsert
+
+/*
+ SYSTEM TABLES
+*/
export const migrations = sqliteTable('migrations', {
name: text('name').notNull().primaryKey(),
@@ -57,12 +73,15 @@ export const migrations = sqliteTable('migrations', {
export type Migration = typeof migrations.$inferSelect
export type NewMigration = typeof migrations.$inferInsert
+export const OPERATIONS = ['INSERT', 'DELETE', 'UPDATE'] as const
+export type Operation = (typeof OPERATIONS)[number]
+
export const changeLog = sqliteTable('change_log', {
id: integer('id', { mode: 'number' })
.primaryKey({ autoIncrement: true })
.notNull(),
tbl_name: text('tbl_name').notNull(),
- op_type: text('op_type').notNull(),
+ op_type: text('op_type', { enum: OPERATIONS }).notNull(),
pk_json: text('pk_json').notNull(),
row_json: text('row_json').notNull(),
changed_at: text('changed_at')
@@ -71,13 +90,3 @@ export const changeLog = sqliteTable('change_log', {
})
export type ChangeLog = typeof changeLog.$inferSelect
export type NewChangeLog = typeof changeLog.$inferInsert
-
-export const cells = sqliteTable(
- 'cells',
- {
- x: integer('x', { mode: 'number' }).notNull(),
- y: integer('y', { mode: 'number' }).notNull(),
- alive: integer('alive', { mode: 'boolean' }).default(false).notNull()
- },
- table => [primaryKey({ columns: [table.x, table.y] })]
-)
diff --git a/src/sqlite/triggers.ts b/src/sqlite/triggers.ts
index 762fd0c..6982afb 100644
--- a/src/sqlite/triggers.ts
+++ b/src/sqlite/triggers.ts
@@ -1,15 +1,19 @@
import type { Client } from '@libsql/client'
+import {
+ type SQLiteTable,
+ type TableConfig,
+ getTableConfig
+} from 'drizzle-orm/sqlite-core'
import * as schema from '../sqlite/schema'
-import { type SQLiteTable, getTableConfig } from 'drizzle-orm/sqlite-core'
-function isSqliteTable(t: any): t is SQLiteTable {
+function isSqliteTable(t: any): t is SQLiteTable {
return t && typeof t.getSQL === 'function' && Boolean(t.getSQL())
}
export async function generateAllTriggers(client: Client) {
- const tables = Object.values(schema).filter(isSqliteTable)
-
+ const tables = Object.values(schema)
for (const tbl of tables) {
+ if (!isSqliteTable(tbl)) continue
const { name: tblName, columns, primaryKeys } = getTableConfig(tbl)
if (tblName === 'change_log') continue
diff --git a/src/sqlite/viewDbWorker.ts b/src/sqlite/viewDbWorker.ts
deleted file mode 100644
index e69de29..0000000
diff --git a/src/sync/cursor.ts b/src/sync/cursor.ts
new file mode 100644
index 0000000..126673b
--- /dev/null
+++ b/src/sync/cursor.ts
@@ -0,0 +1,12 @@
+import { desc } from 'drizzle-orm'
+import type { SqliteRemoteDatabase } from 'drizzle-orm/sqlite-proxy'
+import * as schema from '../sqlite/schema.ts'
+
+export const getLatestId = async (db: SqliteRemoteDatabase) => {
+ const res = await db
+ .select({ id: schema.changeLog.id })
+ .from(schema.changeLog)
+ .orderBy(desc(schema.changeLog.id))
+ .limit(1)
+ return res?.[0]?.id ?? 0
+}
diff --git a/src/sync/socket.ts b/src/sync/socket.ts
new file mode 100644
index 0000000..c3761d4
--- /dev/null
+++ b/src/sync/socket.ts
@@ -0,0 +1,29 @@
+import type { SqliteRemoteDatabase } from 'drizzle-orm/sqlite-proxy'
+import { getLatestId } from './cursor'
+import * as schema from '../sqlite/schema'
+
+export async function setupWebsocket(db: SqliteRemoteDatabase) {
+ const ws = new WebSocket('ws://localhost:3000/ws')
+
+ ws.addEventListener('open', async () => {
+ ws.send(JSON.stringify({ type: 'ping', cursor: await getLatestId(db) }))
+ })
+
+ ws.addEventListener('message', async ({ data }) => {
+ const { type, cursor } = JSON.parse(data)
+
+ if (type === 'pong') {
+ console.log('Server at', cursor, "I'm at", await getLatestId(db))
+ }
+
+ if (type === 'request_data') {
+ ws.send(
+ JSON.stringify({
+ type: 'data',
+ cursor: await getLatestId(db),
+ payload: {}
+ })
+ )
+ }
+ })
+}
diff --git a/src/utils/createBatchFlusher.tsx b/src/utils/createBatchFlusher.tsx
new file mode 100644
index 0000000..cefd3fc
--- /dev/null
+++ b/src/utils/createBatchFlusher.tsx
@@ -0,0 +1,67 @@
+export function createBatchFlusher(
+ sendFn: (items: T[]) => Promise,
+ maxDelay = 5000
+): (items: T[]) => void {
+ let buffer: T[] = []
+ let timer: ReturnType | null = null
+ let inFlight = false
+ let pending = false
+
+ async function flush() {
+ if (inFlight) {
+ pending = true
+ return
+ }
+
+ if (buffer.length === 0) {
+ timer = null
+ return
+ }
+
+ inFlight = true
+ const itemsToSend = buffer
+ buffer = []
+ timer = null
+
+ try {
+ await sendFn(itemsToSend)
+ } catch (error) {
+ console.error('Batch flush failed, retrying:', error)
+
+ buffer = itemsToSend.concat(buffer)
+ if (!timer) timer = setTimeout(flush, maxDelay)
+ } finally {
+ inFlight = false
+
+ if (pending) {
+ pending = false
+ flush()
+ }
+ }
+ }
+
+ return (items: T[]) => {
+ if (items.length === 0) return
+ buffer.push(...items)
+ if (!timer) {
+ timer = setTimeout(flush, maxDelay)
+ }
+ }
+}
+
+export function chainify(
+ fn: (...args: Args) => R | Promise
+): (...args: Args) => Promise {
+ let last: Promise = Promise.resolve()
+
+ return (...args: Args): Promise => {
+ const result = last.then(
+ () => fn(...args),
+ () => fn(...args)
+ ) as Promise
+
+ last = result.catch(e => console.log('error during func chain', e))
+
+ return result
+ }
+}