remove electric sql

Author: letstri

.github/workflows/lint-check.yml

@@ -29,7 +29,7 @@ jobs:
       - name: Setup pnpm
         uses: pnpm/action-setup@v4
         with:
-          version: 11.5.2
+          version: 11.8.0
 
       - name: Install dependencies
         run: pnpm install --frozen-lockfile --ignore-scripts

.github/workflows/release.yml

@@ -19,7 +19,7 @@ jobs:
       - name: Install pnpm
         uses: pnpm/action-setup@v4
         with:
-          version: 11.5.2
+          version: 11.8.0
 
       - name: Install Dependencies
         run: pnpm install --frozen-lockfile

apps/api/.env.example

@@ -30,5 +30,3 @@ INFISICAL_CLIENT_SECRET=
 INFISICAL_ENVIRONMENT=dev
 INFISICAL_PROJECT_ID=
 INFISICAL_SITE_URL=http://localhost:8082
-ELECTRIC_URL=http://localhost:3100
-ELECTRIC_SECRET=devfe1e207b8cfb185b406438a42adev

apps/api/env.ts

@@ -33,8 +33,6 @@ const envType = type({
   GITHUB_TOKEN: 'string',
   TODESKTOP_WEBHOOK_SECRET: 'string',
   PROXY_SHARED_SECRET: 'string',
-  ELECTRIC_URL: 'string',
-  ELECTRIC_SECRET: 'string',
 })
 
 const devOptionalEnvs = [

apps/api/index.ts

@@ -11,9 +11,7 @@ import { createContext } from './orpc/context'
 import { router } from './orpc/routers'
 import { sendEmail } from './lib/resend'
 import { sanitizeLogData } from '@conar/shared/utils/sanitize-log'
-import { ELECTRIC_EXPOSED_HEADERS } from './lib/electric'
 import { healthRouter } from './routers/health'
-import { shapesRouter } from './routers/shapes'
 
 const handler = new RPCHandler(router, {
   interceptors: [
@@ -71,7 +69,6 @@ const app = new Hono<{
       return origin.endsWith('.conar.app') || allowedOrigins.includes(origin) ? origin : null
     },
     credentials: true,
-    exposeHeaders: [...ELECTRIC_EXPOSED_HEADERS],
   }))
   .get('/', c => c.redirect(env.MAIN_URL))
   .use('*', async (c, next) => {
@@ -150,7 +147,6 @@ const app = new Hono<{
 
     return auth.handler(req)
   })
-  .route('/shapes', shapesRouter)
   .use('/rpc/*', async (c, next) => {
     const { matched, response } = await handler.handle(c.req.raw.clone(), {
       prefix: '/rpc',
@@ -168,11 +164,4 @@ const app = new Hono<{
 export default {
   fetch: app.fetch,
   port: Number(process.env.PORT || 3000),
-  // Electric live shape requests (`live=true`) long-poll: Electric holds the
-  // connection open (~20 s) until new data arrives or the poll window elapses.
-  // Bun's default 30 s idle timeout can close these mid-poll, so portless sees
-  // a reset and serves its 502 page. Raise the ceiling above the poll window
-  // (255 s is Bun's max) so legitimate polls survive while truly stuck sockets
-  // still time out.
-  idleTimeout: 255,
 }

apps/api/lib/electric.ts

@@ -20,6 +20,7 @@ export function createContext(c: HonoContext<{ Variables: AppVariables }>) {
   return {
     request: c.req.raw,
     headers: c.req.raw.headers,
+    clientId: c.req.header('x-client-id'),
     setHeader: (key: string, value: string) => {
       c.res.headers.set(key, value)
     },

apps/api/orpc/context.ts

@@ -14,7 +14,7 @@ export const orpc = os.$context<Context>()
 
 export const getUserSecret = memoize((userId: string) => {
   return infisical.secrets.get({ path: ['users', userId], name: INFISICAL_USER_ENCRYPTION_SECRET_NAME })
-}, { maxAge: 10 * 60 * 1000 }) // 10 minutes
+}, { maxAge: 5 * 60 * 1000 }) // 5 minutes
 
 async function getSession(headers: Headers) {
   const session = await auth.api.getSession({ headers })
@@ -29,7 +29,7 @@ async function getSession(headers: Headers) {
 export const logMiddleware = orpc.middleware(async ({ context, next }, input) => {
   const result = await next()
 
-  if (!context.request.url.endsWith('/sync') && !context.request.url.includes('/shapes/')) {
+  if (!context.request.url.endsWith('/sync') && !context.request.url.endsWith('/resolveConnectionString')) {
     context.addLogData({
       input,
       output: (Array.isArray(result.output) && result.output.length > 0)

apps/api/orpc/index.ts

@@ -1,5 +1,9 @@
 import type { Type } from 'arktype'
+import { IORedisPublisher } from '@orpc/experimental-publisher/ioredis'
+import { eventIterator } from '@orpc/server'
 import { type } from 'arktype'
+import { redis } from '~/lib/redis'
+import { authMiddleware, orpc } from '~/orpc'
 
 export function createSyncOutputSchema<const T>(
   schema: type.validate<T>,
@@ -16,6 +20,17 @@ export function createSyncOutputSchema(schema: Type) {
   )
 }
 
+export function createSyncPublisher<T extends Type<{ type: 'insert' | 'update' | 'delete' }>>(
+  output: T,
+  prefix: string,
+) {
+  return new IORedisPublisher<Record<string, typeof output.infer>>({
+    commander: redis.duplicate(),
+    listener: redis.duplicate(),
+    prefix,
+  })
+}
+
 export async function syncDiff<TItem>(opts: {
   input: { id: string, updatedAt: Date }[]
   queries: {
@@ -33,3 +48,15 @@ export async function syncDiff<TItem>(opts: {
   const missingIds = inputIds.filter(id => !existingIds.includes(id))
   return { updatedItems, newItems, missingIds }
 }
+
+// eslint-disable-next-line ts/no-explicit-any
+export function createEventsEndpoint<O extends Type<{ type: 'insert' | 'update' | 'delete' }>, P extends IORedisPublisher<any>>(output: O, publisher: P) {
+  return orpc
+    .use(authMiddleware)
+    .output(eventIterator(output))
+    .handler(async function* ({ context, signal, lastEventId }) {
+      for await (const payload of publisher.subscribe(context.user.id, { signal, lastEventId })) {
+        yield payload
+      }
+    })
+}

apps/api/orpc/lib/sync.ts

@@ -3,17 +3,14 @@ import { chats, chatsMessages, chatsMessagesInsertSchema } from '@conar/db/schem
 import { ORPCError } from '@orpc/server'
 import { type } from 'arktype'
 import { and, eq, inArray } from 'drizzle-orm'
-import { generateTxId } from '~/lib/electric'
 import { orpc, subscriptionMiddleware } from '~/orpc'
+import { publisher } from './events'
 
 const schema = chatsMessagesInsertSchema
 
 export const create = orpc
   .use(subscriptionMiddleware)
-  .input(type.or(
-    schema,
-    schema.array(),
-  ).pipe(data => Array.isArray(data) ? data : [data]))
+  .input(type.or(schema, schema.array()).pipe(data => Array.isArray(data) ? data : [data]))
   .handler(async ({ context, input }) => {
     const chatIds = input.map(item => item.chatId)
     const foundChats = await db.select({ id: chats.id })
@@ -26,9 +23,12 @@ export const create = orpc
       })
     }
 
-    return db.transaction(async (tx) => {
-      await tx.insert(chatsMessages).values(input)
+    const inserted = await db.insert(chatsMessages).values(input).returning()
 
-      return { txid: await generateTxId(tx) }
-    })
+    for (const message of inserted) {
+      publisher.publish(context.user.id, {
+        type: 'insert',
+        value: message,
+      })
+    }
   })