ITO-448: Create Interactions Directly from transcribe stream

lib/auth/events.ts

@@ -317,10 +317,9 @@ export const handleLogin = (
   if (accessToken) {
     mainStore.set(STORE_KEYS.ACCESS_TOKEN, accessToken)
     grpcClient.setAuthToken(accessToken)
-    syncService.start()
   }
 
-  // For self-hosted users, we don't start sync service since they don't have tokens
+  syncService.start()
 }
 
 export const handleLogout = () => {

lib/clients/grpcClient.ts

@@ -474,16 +474,6 @@ class GrpcClient {
   }
 
   async getAdvancedSettings(): Promise<AdvancedSettingsPb | null> {
-    // Check if user is self-hosted and skip server sync
-    const userId = getCurrentUserId()
-    const isSelfHosted = userId === 'self-hosted'
-
-    if (isSelfHosted) {
-      console.log('Self-hosted user detected, using local advanced settings')
-      // Return null for self-hosted users since they don't sync with server
-      return null
-    }
-
     return this.withRetry(async () => {
       const request = create(GetAdvancedSettingsRequestSchema, {})
       return await this.client.getAdvancedSettings(request, {

lib/main/main.ts

@@ -29,7 +29,6 @@ import { initializeMicrophoneSelection } from '../media/microphoneSetUp'
 import { validateStoredTokens, ensureValidTokens } from '../auth/events'
 import { Auth0Config, validateAuth0Config } from '../auth/config'
 import { createAppTray } from './tray'
-import { itoSessionManager } from './itoSessionManager'
 import { initializeAutoUpdater } from './autoUpdaterWrapper'
 import { teardown } from './teardown'
 import { ITO_ENV } from './env'
@@ -84,10 +83,11 @@ app.whenReady().then(async () => {
       | undefined
     if (accessToken) {
       grpcClient.setAuthToken(accessToken)
-      syncService.start()
     }
   }
 
+  syncService.start()
+
   // Setup protocol handling for deep links
   setupProtocolHandling()
 

lib/main/syncService.ts

@@ -9,8 +9,6 @@ import { Note, Interaction, DictionaryItem } from './sqlite/models'
 import mainStore from './store'
 import { STORE_KEYS } from '../constants/store-keys'
 import type { AdvancedSettings } from './store'
-import { DEFAULT_ADVANCED_SETTINGS } from '../constants/generated-defaults.js'
-import { main } from 'bun'
 import { mainWindow } from './app'
 
 const LAST_SYNCED_AT_KEY = 'lastSyncedAt'
@@ -91,7 +89,6 @@ export class SyncService {
       // =================================================================
       let processedChanges = 0
       processedChanges += await this.pushNotes(lastSyncedAt)
-      processedChanges += await this.pushInteractions(lastSyncedAt)
       processedChanges += await this.pushDictionaryItems(lastSyncedAt)
 
       // =================================================================
@@ -138,27 +135,6 @@ export class SyncService {
     return modifiedNotes.length
   }
 
-  private async pushInteractions(lastSyncedAt: string): Promise<number> {
-    const modifiedInteractions =
-      await InteractionsTable.findModifiedSince(lastSyncedAt)
-    if (modifiedInteractions.length > 0) {
-      for (const interaction of modifiedInteractions) {
-        try {
-          if (new Date(interaction.created_at) > new Date(lastSyncedAt)) {
-            await grpcClient.createInteraction(interaction)
-          } else if (interaction.deleted_at) {
-            await grpcClient.deleteInteraction(interaction)
-          } else {
-            await grpcClient.updateInteraction(interaction)
-          }
-        } catch (e) {
-          console.error(`Failed to push interaction ${interaction.id}:`, e)
-        }
-      }
-    }
-    return modifiedInteractions.length
-  }
-
   private async pushDictionaryItems(lastSyncedAt: string): Promise<number> {
     const modifiedItems = await DictionaryTable.findModifiedSince(lastSyncedAt)
     if (modifiedItems.length > 0) {
@@ -240,7 +216,7 @@ export class SyncService {
           created_at: remoteInteraction.createdAt,
           updated_at: remoteInteraction.updatedAt,
           deleted_at: remoteInteraction.deletedAt || null,
-          raw_audio_id: remoteInteraction.rawAudioId,
+          raw_audio_id: remoteInteraction.rawAudioId || null,
           sample_rate: null,
         }
         await InteractionsTable.upsert(localInteraction)
@@ -277,7 +253,6 @@ export class SyncService {
       // Get remote advanced settings
       const remoteSettings = await grpcClient.getAdvancedSettings()
       if (!remoteSettings) {
-        console.warn('No remote advanced settings found, skipping sync.')
         return
       }
 

server/.env.example

@@ -21,7 +21,7 @@ CEREBRAS_API_KEY="your_cerebras_api_key"
 BLOB_STORAGE_BUCKET=ito-blob-storage
 TIMING_BUCKET=ito-timing-storage
 AWS_REGION=us-east-1
-S3_ENDPOINT=http://localhost:9000
+S3_ENDPOINT=http://minio:9000
 S3_ACCESS_KEY_ID=minioadmin
 S3_SECRET_ACCESS_KEY=minioadmin
 S3_FORCE_PATH_STYLE=true

server/docker-compose.yml

@@ -12,7 +12,10 @@ services:
     container_name: ito-server
     restart: always
     depends_on:
-      - db
+      db:
+        condition: service_started
+      createbuckets:
+        condition: service_completed_successfully
     environment:
       NODE_ENV: development
 
@@ -44,9 +47,31 @@ services:
       - minio_data:/data
     healthcheck:
       test: ['CMD', 'mc', 'ready', 'local']
-      interval: 30s
-      timeout: 20s
-      retries: 3
+      interval: 5s
+      timeout: 5s
+      retries: 5
+
+  createbuckets:
+    image: minio/mc:latest
+    container_name: ito-minio-init
+    depends_on:
+      minio:
+        condition: service_healthy
+    entrypoint: >
+      /bin/sh -c "
+      /usr/bin/mc alias set myminio http://minio:9000 $${S3_ACCESS_KEY_ID} $${S3_SECRET_ACCESS_KEY};
+      /usr/bin/mc mb myminio/$${BLOB_STORAGE_BUCKET} --ignore-existing;
+      /usr/bin/mc mb myminio/$${TIMING_BUCKET} --ignore-existing;
+      /usr/bin/mc anonymous set download myminio/$${BLOB_STORAGE_BUCKET};
+      /usr/bin/mc anonymous set download myminio/$${TIMING_BUCKET};
+      echo 'Buckets created successfully';
+      exit 0;
+      "
+    environment:
+      S3_ACCESS_KEY_ID: '$S3_ACCESS_KEY_ID'
+      S3_SECRET_ACCESS_KEY: '$S3_SECRET_ACCESS_KEY'
+      BLOB_STORAGE_BUCKET: '$BLOB_STORAGE_BUCKET'
+      TIMING_BUCKET: '$TIMING_BUCKET'
 
 volumes:
   pgdata:

server/src/auth/authHelpers.ts

@@ -0,0 +1,30 @@
+import type { FastifyRequest } from 'fastify'
+
+/**
+ * Extracts the user ID from a Fastify request.
+ * In production (requireAuth=true), returns the authenticated user's sub from JWT.
+ * In dev mode (requireAuth=false), returns 'self-hosted' as a default user ID.
+ *
+ * @param request - The Fastify request object
+ * @param requireAuth - Whether authentication is required (from REQUIRE_AUTH env var)
+ * @returns The user ID, or undefined if not authenticated and auth is required
+ */
+export function getUserIdFromRequest(
+  request: FastifyRequest,
+  requireAuth: boolean,
+): string | undefined {
+  // Try to get authenticated user ID from JWT
+  const authenticatedUserId = (request as any).user?.sub
+
+  if (authenticatedUserId) {
+    return authenticatedUserId
+  }
+
+  // In dev mode (auth disabled), use 'self-hosted' as default user
+  if (!requireAuth) {
+    return 'self-hosted'
+  }
+
+  // Auth required but no user found
+  return undefined
+}

server/src/db/repo.ts

@@ -11,7 +11,6 @@ import {
 import {
   CreateNoteRequest,
   UpdateNoteRequest,
-  CreateInteractionRequest,
   UpdateInteractionRequest,
   CreateDictionaryItemRequest,
   UpdateDictionaryItemRequest,
@@ -98,12 +97,15 @@ export class NotesRepository {
 }
 
 export class InteractionsRepository {
-  static async create(
-    interactionData: Omit<CreateInteractionRequest, 'rawAudio'> & {
-      userId: string
-      rawAudioId?: string
-    },
-  ): Promise<Interaction> {
+  static async create(interactionData: {
+    id: string
+    userId: string
+    title: string
+    asrOutput: string
+    llmOutput: string | null
+    rawAudioId?: string
+    durationMs: number
+  }): Promise<Interaction> {
     const res = await pool.query<Interaction>(
       `INSERT INTO interactions (id, user_id, title, asr_output, llm_output, raw_audio_id, duration_ms)
        VALUES ($1, $2, $3, $4, $5, $6, $7)
@@ -337,7 +339,6 @@ export class AdvancedSettingsRepository {
     )
 
     const llmSettings = res.rows[0]
-    console.log('Upserted advanced settings:', llmSettings)
     return {
       id: llmSettings.id,
       user_id: llmSettings.user_id,

server/src/server.ts

@@ -170,6 +170,10 @@ export const startServer = async () => {
         if (REQUIRE_AUTH && request.user && request.user.sub) {
           return createContextValues().set(kUser, request.user)
         }
+        // In dev mode (auth disabled), use a self-hosted user
+        if (!REQUIRE_AUTH) {
+          return createContextValues().set(kUser, { sub: 'self-hosted' })
+        }
         return createContextValues()
       },
     })

server/src/services/billing.ts

@@ -5,6 +5,7 @@ import {
   getAuth0ManagementToken,
   getUserInfoFromAuth0,
 } from '../auth/auth0Helpers.js'
+import { getUserIdFromRequest } from '../auth/authHelpers.js'
 
 type Options = {
   requireAuth: boolean
@@ -40,7 +41,7 @@ export const registerBillingRoutes = async (
   fastify.post('/billing/checkout', async (request, reply) => {
     console.log('billing/checkout', request.body)
     try {
-      const userSub = (requireAuth && (request as any).user?.sub) || undefined
+      const userSub = getUserIdFromRequest(request, requireAuth)
       if (!userSub) {
         reply.code(401).send({ success: false, error: 'Unauthorized' })
         return
@@ -79,7 +80,7 @@ export const registerBillingRoutes = async (
 
   fastify.post('/billing/confirm', async (request, reply) => {
     try {
-      const userSub = (requireAuth && (request as any).user?.sub) || undefined
+      const userSub = getUserIdFromRequest(request, requireAuth)
       if (!userSub) {
         reply.code(401).send({ success: false, error: 'Unauthorized' })
         return
@@ -175,7 +176,7 @@ export const registerBillingRoutes = async (
 
   fastify.post('/billing/cancel', async (request, reply) => {
     try {
-      const userSub = (requireAuth && (request as any).user?.sub) || undefined
+      const userSub = getUserIdFromRequest(request, requireAuth)
       if (!userSub) {
         reply.code(401).send({ success: false, error: 'Unauthorized' })
         return
@@ -236,7 +237,7 @@ export const registerBillingRoutes = async (
 
   fastify.get('/billing/status', async (request, reply) => {
     try {
-      const userSub = (requireAuth && (request as any).user?.sub) || undefined
+      const userSub = getUserIdFromRequest(request, requireAuth)
       if (!userSub) {
         reply.code(401).send({ success: false, error: 'Unauthorized' })
         return
@@ -323,7 +324,7 @@ export const registerBillingRoutes = async (
 
   fastify.post('/billing/reactivate', async (request, reply) => {
     try {
-      const userSub = (requireAuth && (request as any).user?.sub) || undefined
+      const userSub = getUserIdFromRequest(request, requireAuth)
       if (!userSub) {
         reply.code(401).send({ success: false, error: 'Unauthorized' })
         return

server/src/services/ito/interactionHelpers.ts

@@ -0,0 +1,66 @@
+import { v4 as uuidv4 } from 'uuid'
+import { getStorageClient } from '../../clients/s3storageClient.js'
+import { createAudioKey } from '../../constants/storage.js'
+import { InteractionsRepository } from '../../db/repo.js'
+import type { Interaction } from '../../db/models.js'
+
+export interface CreateInteractionParams {
+  id: string
+  userId: string
+  title: string
+  asrOutput: string
+  llmOutput: string | null
+  durationMs: number
+  rawAudio?: Buffer
+}
+
+/**
+ * Creates an interaction in the database, optionally uploading audio to S3.
+ * This helper is shared between the gRPC createInteraction endpoint and
+ * the transcribeStreamV2Handler.
+ */
+export async function createInteractionWithAudio(
+  params: CreateInteractionParams,
+): Promise<Interaction> {
+  const { id, userId, title, asrOutput, llmOutput, durationMs, rawAudio } =
+    params
+
+  let rawAudioId: string | undefined
+
+  // If raw audio is provided, upload to S3
+  if (rawAudio && rawAudio.length > 0) {
+    const storageClient = getStorageClient()
+    rawAudioId = uuidv4()
+    const audioKey = createAudioKey(userId, rawAudioId)
+
+    await storageClient.uploadObject(
+      audioKey,
+      rawAudio,
+      undefined, // ContentType
+      {
+        userId,
+        interactionId: id,
+        timestamp: new Date().toISOString(),
+      },
+    )
+
+    console.log(
+      `✅ [${new Date().toISOString()}] Uploaded audio to S3: ${audioKey}`,
+    )
+  }
+
+  // Create interaction in database
+  const interaction = await InteractionsRepository.create({
+    id,
+    userId,
+    title,
+    asrOutput,
+    llmOutput,
+    rawAudioId,
+    durationMs,
+  })
+
+  console.log(`✅ [${new Date().toISOString()}] Created interaction: ${id}`)
+
+  return interaction
+}