feat(desktop): sync knowledge, recordings and shared settings; play any recording
Knowledge documents travel as source-text chunks; each surface embeds them with its own model, the server index is requested through embed-chunks, and documents from the phone are stored without a file and indexed from their chunks. Chunk text is now kept when local embedding fails, so reindexing no longer needs the original file. Recordings upload to the mobile storage contract (audio bucket under the user's folder plus an audio_files row, 50 MiB cap, a Settings > Cloud toggle) and are removed with their record. The history card gains a play button that uses the local file or, for phone recordings, a signed URL. Language (ko/en), system/light/dark theme, auto-polish and the active user command follow the phone's user_settings with its revision rule; changes that arrive from the phone reach the open window.
This commit is contained in:
Yun Chan
parent
cee4ab9317
commit
9a8f7e6aa6
34 changed files with 1405 additions and 64 deletions
|
|
@ -6,6 +6,9 @@ import { IPC_CHANNELS } from '@d3ro/core/ipc-channels'
|
|||
import { ipcSuccess, ipcError, ErrorCode, D3ROError } from '@d3ro/core/errors'
|
||||
import { getCloudSyncService } from '../services/CloudSyncService'
|
||||
import { getLogger } from '../services/LoggerService'
|
||||
import { onConfigChanged } from '../services/ConfigService'
|
||||
import { isApplyingRemoteSettings } from '../services/sync/settings-sync'
|
||||
import { reapplyThemeToAllPopups } from '../windows/WindowManager'
|
||||
|
||||
const logger = getLogger('cloud-sync-handlers')
|
||||
|
||||
|
|
@ -115,6 +118,13 @@ export function registerCloudSyncHandlers(): void {
|
|||
broadcast(IPC_CHANNELS.CLOUD_SYNC.SYNC_ERROR, payload)
|
||||
})
|
||||
// 다른 기기의 변경이 반영되면 앱 공통 새로고침 신호로 알린다 (이력·사전·회의·명령 화면이 구독).
|
||||
// 동기화가 메인에서 바꾼 설정(언어·테마 등)을 화면에도 알린다 — IPC로 바꿀 때만 방송되던 경로.
|
||||
onConfigChanged((event) => {
|
||||
if (!isApplyingRemoteSettings()) return
|
||||
broadcast(IPC_CHANNELS.CONFIG.CHANGED, event)
|
||||
if (event.key === 'theme') reapplyThemeToAllPopups()
|
||||
})
|
||||
|
||||
sync.on('data-changed', (payload) => {
|
||||
broadcast(IPC_CHANNELS.APP.DATA_CHANGED, { type: 'cloud-sync', entities: payload.entities })
|
||||
})
|
||||
|
|
|
|||
|
|
@ -3,15 +3,22 @@
|
|||
import { ipcMain } from 'electron'
|
||||
import { IPC_CHANNELS } from '@d3ro/core/ipc-channels'
|
||||
import { ipcSuccess, ipcError, ErrorCode } from '@d3ro/core/errors'
|
||||
import fs from 'fs'
|
||||
import { getHistoryService } from '../services/HistoryService'
|
||||
import { getCloudSyncService } from '../services/CloudSyncService'
|
||||
import { audioMimeType } from '../services/sync/audio-sync'
|
||||
import type {
|
||||
HistoryQueryParams,
|
||||
HistoryGetByIdParams,
|
||||
HistoryDeleteParams,
|
||||
HistorySearchParams,
|
||||
HistorySetFavoriteParams
|
||||
HistorySetFavoriteParams,
|
||||
HistoryAudioSource
|
||||
} from '@d3ro/core/types'
|
||||
|
||||
/** 렌더러로 넘기는 로컬 녹음 상한. 넘으면 재생하지 않는다(IPC 복사 비용). */
|
||||
const MAX_LOCAL_AUDIO_BYTES = 100 * 1024 * 1024
|
||||
|
||||
export function registerHistoryHandlers(): void {
|
||||
ipcMain.handle(IPC_CHANNELS.HISTORY.GET_ALL, async (_event, params: HistoryQueryParams) => {
|
||||
try {
|
||||
|
|
@ -67,6 +74,27 @@ export function registerHistoryHandlers(): void {
|
|||
}
|
||||
})
|
||||
|
||||
ipcMain.handle(IPC_CHANNELS.HISTORY.GET_AUDIO, async (_event, params: HistoryGetByIdParams) => {
|
||||
try {
|
||||
const entry = getHistoryService().getById(params.id)
|
||||
if (!entry) return ipcError(ErrorCode.HistoryNotFound, `History entry not found: ${params.id}`)
|
||||
const localPath = entry.audioLocalPath
|
||||
if (localPath && fs.existsSync(localPath) && fs.statSync(localPath).size <= MAX_LOCAL_AUDIO_BYTES) {
|
||||
const source: HistoryAudioSource = {
|
||||
kind: 'local',
|
||||
bytes: new Uint8Array(fs.readFileSync(localPath)),
|
||||
mimeType: audioMimeType(localPath) ?? 'audio/wav',
|
||||
}
|
||||
return ipcSuccess(source)
|
||||
}
|
||||
const url = await getCloudSyncService().createHistoryAudioUrl(params.id)
|
||||
return ipcSuccess(url ? ({ kind: 'remote', url } satisfies HistoryAudioSource) : null)
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error)
|
||||
return ipcError(ErrorCode.DBQueryFailed, `Failed to load audio: ${message}`)
|
||||
}
|
||||
})
|
||||
|
||||
ipcMain.handle(IPC_CHANNELS.HISTORY.SEARCH, async (_event, params: HistorySearchParams) => {
|
||||
try {
|
||||
return ipcSuccess(getHistoryService().search(params))
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ import {
|
|||
type RealtimeChannel
|
||||
} from '@supabase/supabase-js'
|
||||
import { getLogger } from './LoggerService'
|
||||
import { configGet, configSet } from './ConfigService'
|
||||
import { configGet, configSet, onConfigChanged } from './ConfigService'
|
||||
import { SUPABASE_URL, SUPABASE_ANON_KEY } from '@d3ro/core/supabase-config'
|
||||
import { openForUser, openLocal, closeCurrent, getCurrentUserId, importLocalModeData } from '../db'
|
||||
import { D3ROError, ErrorCode } from '@d3ro/core/errors'
|
||||
|
|
@ -24,6 +24,8 @@ import { enqueueChange, getSyncState, setSyncState } from './sync/sync-outbox'
|
|||
import { checkInDesktopDevice, currentDeviceInfo, unregisterDesktopDevice } from './sync/device-registration'
|
||||
import type { SyncEntity, SyncRunResult } from './sync/sync-types'
|
||||
import { nodeRealtimeTransport } from './sync/realtime-transport'
|
||||
import { SETTINGS_ROW_ID, SYNCED_CONFIG_KEYS, isApplyingRemoteSettings } from './sync/settings-sync'
|
||||
import { AUDIO_BUCKET, listLocalAudioOwners } from './sync/audio-sync'
|
||||
|
||||
const logger = getLogger('CloudSyncService')
|
||||
|
||||
|
|
@ -46,6 +48,8 @@ const REALTIME_TABLES = [
|
|||
'memo_tags',
|
||||
'custom_instructions',
|
||||
'user_templates',
|
||||
'knowledge_documents',
|
||||
'user_settings',
|
||||
'sync_tombstones',
|
||||
] as const
|
||||
|
||||
|
|
@ -125,6 +129,23 @@ class CloudSyncService extends EventEmitter {
|
|||
|
||||
this._lastSyncAt = (configGet('cloudSyncLastAt') as number | undefined) ?? null
|
||||
|
||||
// 언어·테마·자동 다듬기·활성 명령이 바뀌면 모바일 user_settings 로 올린다(원격 반영 중엔 제외).
|
||||
onConfigChanged((event) => {
|
||||
if (isApplyingRemoteSettings()) return
|
||||
if ((SYNCED_CONFIG_KEYS as readonly string[]).includes(event.key)) {
|
||||
this.pushOne('user_settings', SETTINGS_ROW_ID)
|
||||
}
|
||||
// 녹음 동기화를 켜면 그동안 올리지 않은 녹음을 올린다.
|
||||
if (event.key === 'cloudSyncAudio' && event.value === true && this._engine) {
|
||||
try {
|
||||
for (const id of listLocalAudioOwners('history')) this.pushOne('history_audio', id)
|
||||
for (const id of listLocalAudioOwners('meeting')) this.pushOne('meeting_audio', id)
|
||||
} catch (err) {
|
||||
logger.warn(`Audio re-queue failed: ${err instanceof Error ? err.message : String(err)}`)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
// 저장된 refresh token 복원
|
||||
const stored = this._loadStoredRefreshToken()
|
||||
if (stored) {
|
||||
|
|
@ -913,6 +934,29 @@ class CloudSyncService extends EventEmitter {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 다른 기기(모바일·웹)에서 녹음한 기록의 재생 URL(5분 서명). 로컬 파일이 없을 때만 쓴다.
|
||||
* 로그인 전이거나 서버에 녹음이 없으면 null.
|
||||
*/
|
||||
async createHistoryAudioUrl(historyId: string): Promise<string | null> {
|
||||
const client = this._client
|
||||
const session = this._session
|
||||
if (!client || !session) return null
|
||||
const { data, error } = await client
|
||||
.from('audio_files')
|
||||
.select('storage_key,upload_status')
|
||||
.eq('user_id', session.user.id)
|
||||
.eq('history_id', historyId)
|
||||
.eq('upload_status', 'uploaded')
|
||||
.order('created_at', { ascending: false })
|
||||
.limit(1)
|
||||
.maybeSingle()
|
||||
if (error || !data || typeof data.storage_key !== 'string') return null
|
||||
if (!data.storage_key.startsWith(`${session.user.id}/`)) return null
|
||||
const signed = await client.storage.from(AUDIO_BUCKET).createSignedUrl(data.storage_key, 300)
|
||||
return signed.error ? null : signed.data.signedUrl
|
||||
}
|
||||
|
||||
getSyncStatus(): SyncStatus | null {
|
||||
if (!this._engine) return null
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -158,6 +158,7 @@ const CONFIG_DEFAULTS: AppConfig = {
|
|||
supabaseAnonKey: '',
|
||||
cloudSyncLastAt: null,
|
||||
deviceInstallationId: null,
|
||||
cloudSyncAudio: true,
|
||||
onboardingCompleted: false,
|
||||
// Phase 6/10+: AppConfig 키 기본값 (WS2 SSOT 강화 대응).
|
||||
// as never 제거 후 configGet이 이 키들을 반환 — 기존 사용자 config(0.1.x)에
|
||||
|
|
|
|||
|
|
@ -41,6 +41,7 @@ class HistoryService {
|
|||
|
||||
// Phase 3.3: 로그인 상태면 자동 push (fire-and-forget, 로컬 write는 차단하지 않음)
|
||||
void getCloudSyncService().pushOne('history', id)
|
||||
if (input.audioLocalPath) getCloudSyncService().pushOne('history_audio', id)
|
||||
|
||||
// 비동기로 LLM 타이틀 자동 생성 (fire-and-forget)
|
||||
this.generateTitle(id).catch(() => { /* ignore */ })
|
||||
|
|
|
|||
|
|
@ -369,6 +369,7 @@ class MeetingModeService extends EventEmitter {
|
|||
if (!fs.existsSync(audioDir)) fs.mkdirSync(audioDir, { recursive: true })
|
||||
const wavPath = path.join(audioDir, `${sessionId}.wav`)
|
||||
this._saveWav(wavPath, Buffer.concat(this._audioBuffersForFile))
|
||||
getCloudSyncService().pushOne('meeting_audio', sessionId)
|
||||
logger.info(`회의 오디오 저장: ${wavPath}`)
|
||||
} catch (err) {
|
||||
logger.warn(`오디오 저장 실패: ${err instanceof Error ? err.message : String(err)}`)
|
||||
|
|
|
|||
|
|
@ -4,13 +4,15 @@
|
|||
|
||||
import { EventEmitter } from 'events'
|
||||
import path from 'path'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import fs from 'fs'
|
||||
import { asc, eq } from 'drizzle-orm'
|
||||
import { getLogger } from './LoggerService'
|
||||
import { getPremiumLLMService } from './PremiumLLMService'
|
||||
import { getOllamaServerUrl } from './LocalLLMService'
|
||||
import { getDatabase } from '../db'
|
||||
import { ragDocuments, ragChunks } from '../db/schema'
|
||||
import { getMainWindow } from '../windows/WindowManager'
|
||||
import { getCloudSyncService } from './CloudSyncService'
|
||||
import { IPC_CHANNELS } from '@d3ro/core/ipc-channels'
|
||||
import { D3ROError, ErrorCode } from '@d3ro/core/errors'
|
||||
import type {
|
||||
|
|
@ -133,8 +135,12 @@ class RAGService extends EventEmitter {
|
|||
addedAt: Date.now(),
|
||||
}).run()
|
||||
|
||||
// 청크 원문을 먼저 저장한다 — 임베딩이 실패해도 원문은 남아 재색인·기기 간 동기화가 가능하다.
|
||||
this._storeChunks(docId, chunks)
|
||||
getCloudSyncService().pushOne('knowledge_documents', docId)
|
||||
|
||||
// 비동기 인덱싱 (임베딩 생성)
|
||||
this._indexDocument(docId, fileName, chunks).catch((err) => {
|
||||
this._embedStoredChunks(docId, fileName).catch((err) => {
|
||||
logger.error(`Indexing failed for ${fileName}:`, err)
|
||||
})
|
||||
|
||||
|
|
@ -154,10 +160,59 @@ class RAGService extends EventEmitter {
|
|||
* 문서 제거 (청크 포함)
|
||||
*/
|
||||
removeDocument(documentId: string): void {
|
||||
this.removeRemote(documentId)
|
||||
getCloudSyncService().pushDelete('knowledge_documents', documentId)
|
||||
logger.info(`RAG document removed: ${documentId}`)
|
||||
}
|
||||
|
||||
/** 동기화: 다른 기기에서 지운 문서를 지운다(outbox에 넣지 않는다). */
|
||||
removeRemote(documentId: string): boolean {
|
||||
const db = getDatabase()
|
||||
db.delete(ragChunks).where(eq(ragChunks.documentId, documentId)).run()
|
||||
db.delete(ragDocuments).where(eq(ragDocuments.id, documentId)).run()
|
||||
logger.info(`RAG document removed: ${documentId}`)
|
||||
return db.delete(ragDocuments).where(eq(ragDocuments.id, documentId)).run().changes > 0
|
||||
}
|
||||
|
||||
/**
|
||||
* 동기화: 다른 기기(모바일·웹)의 지식 문서를 원문 청크로 받아 저장하고, 이 기기의 임베딩 모델로 색인한다.
|
||||
* 임베딩 공간이 기기마다 달라 벡터는 옮기지 않는다. 원본 파일은 없으므로 filePath는 비워 둔다.
|
||||
*/
|
||||
applyRemoteDocument(doc: {
|
||||
id: string
|
||||
fileName: string
|
||||
fileType: RAGDocument['fileType']
|
||||
chunks: string[]
|
||||
addedAt: number
|
||||
}): boolean {
|
||||
const db = getDatabase()
|
||||
if (db.select({ id: ragDocuments.id }).from(ragDocuments).where(eq(ragDocuments.id, doc.id)).get()) return false
|
||||
const chunks = doc.chunks.filter((c) => c.trim().length > 0)
|
||||
if (chunks.length === 0) return false
|
||||
db.insert(ragDocuments).values({
|
||||
id: doc.id,
|
||||
fileName: doc.fileName,
|
||||
filePath: '',
|
||||
fileType: doc.fileType,
|
||||
chunkCount: chunks.length,
|
||||
indexed: false,
|
||||
indexedAt: null,
|
||||
addedAt: doc.addedAt,
|
||||
}).run()
|
||||
this._storeChunks(doc.id, chunks)
|
||||
this._embedStoredChunks(doc.id, doc.fileName).catch((err) => {
|
||||
logger.warn(`Synced document ${doc.fileName} is stored but not embedded yet:`, err)
|
||||
})
|
||||
return true
|
||||
}
|
||||
|
||||
/** 저장된 청크 원문(chunkIndex 순) — 동기화 업로드용 */
|
||||
getStoredChunks(documentId: string): string[] {
|
||||
return getDatabase()
|
||||
.select({ content: ragChunks.content })
|
||||
.from(ragChunks)
|
||||
.where(eq(ragChunks.documentId, documentId))
|
||||
.orderBy(asc(ragChunks.chunkIndex))
|
||||
.all()
|
||||
.map((r) => r.content)
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -171,8 +226,11 @@ class RAGService extends EventEmitter {
|
|||
}
|
||||
const doc = rows[0]
|
||||
|
||||
// 기존 청크 삭제
|
||||
db.delete(ragChunks).where(eq(ragChunks.documentId, documentId)).run()
|
||||
// 원본 파일이 없으면(다른 기기에서 동기화된 문서 등) 저장된 원문 청크로 다시 임베딩한다.
|
||||
if (!doc.filePath || !fs.existsSync(doc.filePath)) {
|
||||
await this._embedStoredChunks(documentId, doc.fileName, { force: true })
|
||||
return
|
||||
}
|
||||
|
||||
// 텍스트 재추출 + 재인덱싱
|
||||
const content = await this._extractText(doc.filePath, doc.fileType as RAGDocument['fileType'])
|
||||
|
|
@ -183,7 +241,9 @@ class RAGService extends EventEmitter {
|
|||
.where(eq(ragDocuments.id, documentId))
|
||||
.run()
|
||||
|
||||
await this._indexDocument(documentId, doc.fileName, chunks)
|
||||
this._storeChunks(documentId, chunks)
|
||||
getCloudSyncService().pushOne('knowledge_documents', documentId)
|
||||
await this._embedStoredChunks(documentId, doc.fileName)
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -194,7 +254,8 @@ class RAGService extends EventEmitter {
|
|||
|
||||
try {
|
||||
const db = getDatabase()
|
||||
const allChunks = db.select().from(ragChunks).all()
|
||||
// 원문만 있고 아직 임베딩되지 않은 청크는 검색 대상이 아니다.
|
||||
const allChunks = db.select().from(ragChunks).all().filter((c) => c.embedding.length > 0)
|
||||
if (allChunks.length === 0) {
|
||||
throw new D3ROError(ErrorCode.RAGQueryFailed, 'No indexed chunks to query')
|
||||
}
|
||||
|
|
@ -245,9 +306,37 @@ ${context}`
|
|||
|
||||
// ── 내부 메서드 ──
|
||||
|
||||
private async _indexDocument(docId: string, fileName: string, chunks: string[]): Promise<void> {
|
||||
/** 청크 원문을 (다시) 저장한다. 임베딩은 비워 두고 _embedStoredChunks 가 채운다. */
|
||||
private _storeChunks(docId: string, chunks: string[]): void {
|
||||
const db = getDatabase()
|
||||
db.transaction((tx) => {
|
||||
tx.delete(ragChunks).where(eq(ragChunks.documentId, docId)).run()
|
||||
chunks.forEach((content, index) => {
|
||||
tx.insert(ragChunks).values({
|
||||
id: crypto.randomUUID(),
|
||||
documentId: docId,
|
||||
content,
|
||||
embedding: '',
|
||||
chunkIndex: index,
|
||||
}).run()
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/** 저장된 청크 중 임베딩이 없는 것(force면 전부)을 이 기기의 임베딩 모델로 채운다. */
|
||||
private async _embedStoredChunks(
|
||||
docId: string,
|
||||
fileName: string,
|
||||
options: { force?: boolean } = {}
|
||||
): Promise<void> {
|
||||
this._state = 'indexing'
|
||||
const db = getDatabase()
|
||||
const chunks = db
|
||||
.select()
|
||||
.from(ragChunks)
|
||||
.where(eq(ragChunks.documentId, docId))
|
||||
.orderBy(asc(ragChunks.chunkIndex))
|
||||
.all()
|
||||
|
||||
logger.info(`RAG indexing started: ${fileName} (${chunks.length} chunks)`)
|
||||
|
||||
|
|
@ -262,21 +351,21 @@ ${context}`
|
|||
|
||||
let successCount = 0
|
||||
for (let i = 0; i < chunks.length; i++) {
|
||||
const chunk = chunks[i]
|
||||
if (chunk.embedding.length > 0 && !options.force) {
|
||||
successCount++
|
||||
continue
|
||||
}
|
||||
try {
|
||||
const embedding = await this._embed(chunks[i])
|
||||
|
||||
db.insert(ragChunks).values({
|
||||
id: crypto.randomUUID(),
|
||||
documentId: docId,
|
||||
content: chunks[i],
|
||||
embedding: JSON.stringify(embedding),
|
||||
chunkIndex: i,
|
||||
}).run()
|
||||
|
||||
const embedding = await this._embed(chunk.content)
|
||||
db.update(ragChunks)
|
||||
.set({ embedding: JSON.stringify(embedding) })
|
||||
.where(eq(ragChunks.id, chunk.id))
|
||||
.run()
|
||||
successCount++
|
||||
} catch (err) {
|
||||
logger.warn(`RAG embedding failed for chunk ${i}/${chunks.length} of ${fileName}:`, err)
|
||||
// 개별 청크 실패는 건너뛰고 계속 진행
|
||||
// 개별 청크 실패는 건너뛰고 계속 진행 — 원문은 남아 있어 재색인할 수 있다
|
||||
}
|
||||
|
||||
const progress: RAGIndexProgress = {
|
||||
|
|
@ -307,9 +396,9 @@ ${context}`
|
|||
)
|
||||
}
|
||||
|
||||
// 인덱싱 완료 표시
|
||||
// 인덱싱 완료 표시 (chunkCount는 원문 청크 수 — 서버·모바일과 같은 기준)
|
||||
db.update(ragDocuments)
|
||||
.set({ indexed: true, indexedAt: Date.now(), chunkCount: successCount })
|
||||
.set({ indexed: true, indexedAt: Date.now(), chunkCount: chunks.length })
|
||||
.where(eq(ragDocuments.id, docId))
|
||||
.run()
|
||||
|
||||
|
|
|
|||
|
|
@ -18,6 +18,8 @@ import {
|
|||
pushMemoTagRemovals,
|
||||
reconcileMemoTags,
|
||||
} from './memo-tag-sync'
|
||||
import { fetchRemoteAudioOwners, isAudioSyncEnabled, listLocalAudioOwners, pushAudio } from './audio-sync'
|
||||
import { SETTINGS_ROW_ID, applyRemoteSettings, fetchRemoteSettings, pushSettings } from './settings-sync'
|
||||
import {
|
||||
completeEntry,
|
||||
dropEntry,
|
||||
|
|
@ -215,6 +217,25 @@ export class SyncEngine extends EventEmitter {
|
|||
}
|
||||
}
|
||||
}
|
||||
const ctx: PushContext = { remote: this.remote, userId: this.userId }
|
||||
// 설정: 서버에 행이 없을 때만 로컬 값을 올린다. 있으면 다른 기기의 선택이 우선(pull이 반영).
|
||||
if (!(await fetchRemoteSettings(ctx))) {
|
||||
enqueueChange('user_settings', SETTINGS_ROW_ID, 'upsert', now)
|
||||
queued++
|
||||
}
|
||||
if (isAudioSyncEnabled()) {
|
||||
for (const owner of ['history', 'meeting'] as const) {
|
||||
const local = listLocalAudioOwners(owner)
|
||||
if (local.length === 0) continue
|
||||
const remote = await fetchRemoteAudioOwners(this.remote, this.userId, owner)
|
||||
for (const id of local) {
|
||||
if (!remote.has(id)) {
|
||||
enqueueChange(owner === 'history' ? 'history_audio' : 'meeting_audio', id, 'upsert', now)
|
||||
queued++
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
setSyncState(BACKFILL_FLAG, 'done', now)
|
||||
logger.info(`Backfill queued ${queued} local change(s) for ${this.userId}`)
|
||||
}
|
||||
|
|
@ -269,31 +290,51 @@ export class SyncEngine extends EventEmitter {
|
|||
'meeting_memos',
|
||||
'meeting_documents',
|
||||
'custom_instructions',
|
||||
'user_settings',
|
||||
'user_templates',
|
||||
'knowledge_documents',
|
||||
'memo_tags',
|
||||
'history_audio',
|
||||
'meeting_audio',
|
||||
]
|
||||
let networkDown = false
|
||||
for (const entity of upsertOrder) {
|
||||
const bucket = byEntity.get(entity)
|
||||
if (!bucket || bucket.upserts.length === 0 || networkDown) continue
|
||||
const outcomes =
|
||||
entity === 'memo_tags'
|
||||
? await pushMemoTagAdds(ctx, bucket.upserts.map((e) => e.rowId))
|
||||
: await this.requireAdapter(entity).push(ctx, bucket.upserts.map((e) => e.rowId))
|
||||
const ids = bucket.upserts.map((e) => e.rowId)
|
||||
const outcomes = await this.pushUpserts(ctx, entity, ids)
|
||||
networkDown = this.settle(bucket.upserts, outcomes, result, entity)
|
||||
}
|
||||
// 자식 먼저 delete (서버 cascade가 있어도 순서를 지켜 FK 오류를 피한다)
|
||||
for (const entity of [...order].reverse()) {
|
||||
const bucket = byEntity.get(entity)
|
||||
if (!bucket || bucket.deletes.length === 0 || networkDown) continue
|
||||
const ids = bucket.deletes.map((e) => e.rowId)
|
||||
const outcomes =
|
||||
entity === 'memo_tags'
|
||||
? await pushMemoTagRemovals(ctx, bucket.deletes.map((e) => e.rowId))
|
||||
: await this.requireAdapter(entity).pushDeletes(ctx, bucket.deletes.map((e) => e.rowId))
|
||||
? await pushMemoTagRemovals(ctx, ids)
|
||||
: adapterFor(entity)
|
||||
? await this.requireAdapter(entity).pushDeletes(ctx, ids)
|
||||
: ids.map((id) => ({ id, error: null }))
|
||||
networkDown = this.settle(bucket.deletes, outcomes, result, entity)
|
||||
}
|
||||
}
|
||||
|
||||
private async pushUpserts(ctx: PushContext, entity: SyncEntity, ids: string[]): Promise<PushOutcome[]> {
|
||||
switch (entity) {
|
||||
case 'memo_tags':
|
||||
return pushMemoTagAdds(ctx, ids)
|
||||
case 'user_settings':
|
||||
return [{ id: SETTINGS_ROW_ID, error: await pushSettings(ctx) }]
|
||||
case 'history_audio':
|
||||
return pushAudio(ctx, 'history', ids)
|
||||
case 'meeting_audio':
|
||||
return pushAudio(ctx, 'meeting', ids)
|
||||
default:
|
||||
return this.requireAdapter(entity).push(ctx, ids)
|
||||
}
|
||||
}
|
||||
|
||||
/** outbox를 결과대로 정리. 네트워크 계열 실패가 있었으면 true(이번 flush 중단). */
|
||||
private settle(entries: OutboxEntry[], outcomes: PushOutcome[], result: SyncRunResult, entity: SyncEntity): boolean {
|
||||
const byId = new Map(outcomes.map((o) => [o.id, o.error]))
|
||||
|
|
@ -361,6 +402,20 @@ export class SyncEngine extends EventEmitter {
|
|||
logger.warn(`Reconcile memo_tags failed: ${message}`)
|
||||
}
|
||||
|
||||
try {
|
||||
if (!pendingOps('user_settings').has(SETTINGS_ROW_ID)) {
|
||||
const row = await fetchRemoteSettings({ remote: this.remote, userId: this.userId })
|
||||
if (row && applyRemoteSettings(row)) {
|
||||
changed.add('user_settings')
|
||||
result.pulled++
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err)
|
||||
result.errors.push(`user_settings: ${message}`)
|
||||
logger.warn(`Pull user_settings failed: ${message}`)
|
||||
}
|
||||
|
||||
result.changed = [...changed]
|
||||
if (result.errors.length === 0) setSyncState('lastPullAt', String(this.now()))
|
||||
}
|
||||
|
|
@ -389,7 +444,7 @@ export class SyncEngine extends EventEmitter {
|
|||
const id = typeof row.id === 'string' ? row.id : null
|
||||
if (id && !pending.has(id)) {
|
||||
try {
|
||||
if (adapter.applyRemote(row)) applied++
|
||||
if (await adapter.applyRemote(row, { remote: this.remote, userId: this.userId })) applied++
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
`Apply ${adapter.entity}/${id} failed: ${err instanceof Error ? err.message : String(err)}`
|
||||
|
|
|
|||
219
apps/desktop/src/main/services/sync/audio-sync.ts
Normal file
219
apps/desktop/src/main/services/sync/audio-sync.ts
Normal file
|
|
@ -0,0 +1,219 @@
|
|||
// src/main/services/sync/audio-sync.ts
|
||||
// 데스크톱 녹음 파일을 모바일과 같은 저장소 계약(`audio` 버킷, `{userId}/...` 경로, audio_files 행)으로 올린다.
|
||||
// 모바일은 audio_files 로 재생 URL을 만들고, 데스크톱은 반대로 모바일 녹음을 재생할 때 서명 URL을 받는다.
|
||||
|
||||
import fs from 'fs'
|
||||
import path from 'path'
|
||||
import { createHash } from 'crypto'
|
||||
import { app } from 'electron'
|
||||
import { eq, isNotNull } from 'drizzle-orm'
|
||||
import { getDatabase } from '../../db'
|
||||
import { history, meetingSessions } from '../../db/schema'
|
||||
import { configGet } from '../ConfigService'
|
||||
import { getLogger } from '../LoggerService'
|
||||
import { isUuid, type PushContext } from './sync-adapters'
|
||||
import { toSyncRemoteError, type PushOutcome, type RemoteCursor, type SyncRemote } from './sync-types'
|
||||
|
||||
const logger = getLogger('AudioSync')
|
||||
|
||||
export const AUDIO_BUCKET = 'audio'
|
||||
/** Supabase Storage 파일 한도(config.toml storage 50MiB). 넘는 파일은 올리지 않는다. */
|
||||
export const MAX_AUDIO_BYTES = 50 * 1024 * 1024
|
||||
|
||||
export type AudioOwner = 'history' | 'meeting'
|
||||
|
||||
const OWNER_COLUMN: Record<AudioOwner, 'history_id' | 'meeting_id'> = {
|
||||
history: 'history_id',
|
||||
meeting: 'meeting_id',
|
||||
}
|
||||
|
||||
/** audio 버킷이 허용하는 형식만(서버 allowed_mime_types). 그 밖의 파일은 올리지 않는다. */
|
||||
const AUDIO_MIME: Record<string, string> = {
|
||||
'.wav': 'audio/wav',
|
||||
'.webm': 'audio/webm',
|
||||
'.mp3': 'audio/mpeg',
|
||||
'.m4a': 'audio/x-m4a',
|
||||
'.aac': 'audio/aac',
|
||||
'.ogg': 'audio/ogg',
|
||||
'.flac': 'audio/flac',
|
||||
'.mp4': 'video/mp4',
|
||||
'.mov': 'video/quicktime',
|
||||
}
|
||||
|
||||
export function audioMimeType(filePath: string): string | null {
|
||||
return AUDIO_MIME[path.extname(filePath).toLowerCase()] ?? null
|
||||
}
|
||||
|
||||
export function audioStorageKey(userId: string, owner: AudioOwner, id: string, extension = '.wav'): string {
|
||||
return `${userId}/desktop/${owner}/${id}${extension.toLowerCase()}`
|
||||
}
|
||||
|
||||
export function isAudioSyncEnabled(): boolean {
|
||||
return configGet('cloudSyncAudio') !== false
|
||||
}
|
||||
|
||||
function meetingAudioPath(id: string): string {
|
||||
return path.join(app.getPath('userData'), 'meeting-audio', `${id}.wav`)
|
||||
}
|
||||
|
||||
/** 로컬 녹음 파일 경로. 파일이 없으면 null */
|
||||
export function localAudioPath(owner: AudioOwner, id: string): string | null {
|
||||
let candidate: string | null
|
||||
if (owner === 'history') {
|
||||
const row = getDatabase()
|
||||
.select({ audioLocalPath: history.audioLocalPath })
|
||||
.from(history)
|
||||
.where(eq(history.id, id))
|
||||
.get()
|
||||
candidate = row?.audioLocalPath ?? null
|
||||
} else {
|
||||
candidate = meetingAudioPath(id)
|
||||
}
|
||||
return candidate && fs.existsSync(candidate) ? candidate : null
|
||||
}
|
||||
|
||||
/** 녹음이 있는 로컬 항목 id (최초 대조용) */
|
||||
export function listLocalAudioOwners(owner: AudioOwner): string[] {
|
||||
if (owner === 'history') {
|
||||
return getDatabase()
|
||||
.select({ id: history.id, audioLocalPath: history.audioLocalPath })
|
||||
.from(history)
|
||||
.where(isNotNull(history.audioLocalPath))
|
||||
.all()
|
||||
.filter((r) => isUuid(r.id) && r.audioLocalPath !== null && fs.existsSync(r.audioLocalPath))
|
||||
.map((r) => r.id)
|
||||
}
|
||||
return getDatabase()
|
||||
.select({ id: meetingSessions.id })
|
||||
.from(meetingSessions)
|
||||
.all()
|
||||
.filter((r) => isUuid(r.id) && fs.existsSync(meetingAudioPath(r.id)))
|
||||
.map((r) => r.id)
|
||||
}
|
||||
|
||||
function durationMs(owner: AudioOwner, id: string): number | null {
|
||||
if (owner === 'history') {
|
||||
const row = getDatabase().select({ duration: history.duration }).from(history).where(eq(history.id, id)).get()
|
||||
return row ? Math.round(row.duration * 1000) : null
|
||||
}
|
||||
const row = getDatabase()
|
||||
.select({ durationMs: meetingSessions.durationMs })
|
||||
.from(meetingSessions)
|
||||
.where(eq(meetingSessions.id, id))
|
||||
.get()
|
||||
return row?.durationMs ?? null
|
||||
}
|
||||
|
||||
export async function pushAudio(ctx: PushContext, owner: AudioOwner, ids: string[]): Promise<PushOutcome[]> {
|
||||
const outcomes: PushOutcome[] = []
|
||||
const column = OWNER_COLUMN[owner]
|
||||
for (const id of ids) {
|
||||
const filePath = isAudioSyncEnabled() && isUuid(id) ? localAudioPath(owner, id) : null
|
||||
const mimeType = filePath ? audioMimeType(filePath) : null
|
||||
if (!filePath || !mimeType) {
|
||||
outcomes.push({ id, error: null })
|
||||
continue
|
||||
}
|
||||
try {
|
||||
const size = fs.statSync(filePath).size
|
||||
if (size > MAX_AUDIO_BYTES) {
|
||||
logger.info(`Audio ${owner}/${id} is ${size} bytes — over the storage limit, kept local only`)
|
||||
outcomes.push({ id, error: null })
|
||||
continue
|
||||
}
|
||||
const bytes = fs.readFileSync(filePath)
|
||||
const sha256 = createHash('sha256').update(bytes).digest('hex')
|
||||
const existing = await ctx.remote.selectWhere(
|
||||
'audio_files',
|
||||
ctx.userId,
|
||||
[
|
||||
{ column, op: 'eq', value: id },
|
||||
{ column: 'sha256', op: 'eq', value: sha256 },
|
||||
{ column: 'upload_status', op: 'eq', value: 'uploaded' },
|
||||
],
|
||||
'id'
|
||||
)
|
||||
if (existing.length > 0) {
|
||||
outcomes.push({ id, error: null })
|
||||
continue
|
||||
}
|
||||
const key = audioStorageKey(ctx.userId, owner, id, path.extname(filePath))
|
||||
await ctx.remote.uploadObject(AUDIO_BUCKET, key, new Uint8Array(bytes), mimeType)
|
||||
await ctx.remote.upsert(
|
||||
'audio_files',
|
||||
[
|
||||
{
|
||||
user_id: ctx.userId,
|
||||
[column]: id,
|
||||
source: 'recording',
|
||||
original_name: path.basename(filePath),
|
||||
storage_key: key,
|
||||
mime_type: mimeType,
|
||||
size_bytes: size,
|
||||
duration_ms: durationMs(owner, id),
|
||||
sha256,
|
||||
upload_status: 'uploaded',
|
||||
},
|
||||
],
|
||||
'user_id,sha256,storage_key'
|
||||
)
|
||||
outcomes.push({ id, error: null })
|
||||
} catch (err) {
|
||||
outcomes.push({ id, error: toSyncRemoteError(err) })
|
||||
}
|
||||
}
|
||||
return outcomes
|
||||
}
|
||||
|
||||
/** 서버에 이미 올라간 녹음의 주인 id (최초 대조용) */
|
||||
export async function fetchRemoteAudioOwners(remote: SyncRemote, userId: string, owner: AudioOwner): Promise<Set<string>> {
|
||||
const column = OWNER_COLUMN[owner]
|
||||
const owners = new Set<string>()
|
||||
let after: RemoteCursor | null = null
|
||||
for (;;) {
|
||||
const page = await remote.fetchPage({
|
||||
table: 'audio_files',
|
||||
userId,
|
||||
cursorColumn: 'updated_at',
|
||||
after,
|
||||
limit: 1000,
|
||||
columns: `id,${column},updated_at`,
|
||||
filters: [{ column: 'upload_status', op: 'eq', value: 'uploaded' }],
|
||||
})
|
||||
for (const row of page) {
|
||||
const value = row[column]
|
||||
if (isUuid(value)) owners.add(value)
|
||||
}
|
||||
if (page.length < 1000) break
|
||||
const last = page[page.length - 1]
|
||||
if (typeof last.updated_at !== 'string' || typeof last.id !== 'string') break
|
||||
after = { ts: last.updated_at, id: last.id }
|
||||
}
|
||||
return owners
|
||||
}
|
||||
|
||||
/**
|
||||
* 지운 기록·회의의 녹음을 저장소와 audio_files 에서 지운다.
|
||||
* 부모 행이 지워지면 audio_files 는 SET NULL 로 고아가 되므로 부모보다 먼저 지운다. 실패해도 부모 삭제는 진행한다.
|
||||
*/
|
||||
export async function purgeRemoteAudio(ctx: PushContext, column: 'history_id' | 'meeting_id', ids: string[]): Promise<void> {
|
||||
for (const id of ids.filter(isUuid)) {
|
||||
try {
|
||||
const rows = await ctx.remote.selectWhere(
|
||||
'audio_files',
|
||||
ctx.userId,
|
||||
[{ column, op: 'eq', value: id }],
|
||||
'id,storage_key'
|
||||
)
|
||||
const keys = rows.map((r) => r.storage_key).filter((k): k is string => typeof k === 'string')
|
||||
await ctx.remote.removeObjects(AUDIO_BUCKET, keys)
|
||||
await ctx.remote.deleteByIds(
|
||||
'audio_files',
|
||||
ctx.userId,
|
||||
rows.map((r) => r.id).filter(isUuid)
|
||||
)
|
||||
} catch (err) {
|
||||
logger.warn(`Audio cleanup for ${column}=${id} failed: ${err instanceof Error ? err.message : String(err)}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
169
apps/desktop/src/main/services/sync/settings-sync.ts
Normal file
169
apps/desktop/src/main/services/sync/settings-sync.ts
Normal file
|
|
@ -0,0 +1,169 @@
|
|||
// src/main/services/sync/settings-sync.ts
|
||||
// 모바일 user_settings(한 행, revision 낙관적 동시성)와 데스크톱 설정 중 뜻이 같은 것만 맞춘다.
|
||||
// locale ↔ language (모바일이 받는 ko/en만)
|
||||
// theme_mode ↔ theme (system/light/dark 만 — 데스크톱 전용 테마는 로컬에 둔다)
|
||||
// auto_polish_enabled ↔ defaultLLMAction ('none' 이면 꺼짐)
|
||||
// active_instruction_id ↔ activeInstructionId (양쪽에 있는 사용자 명령일 때만)
|
||||
|
||||
import type { LLMActionSelection, ThemeMode } from '@d3ro/core/types'
|
||||
import { configGet, configSet } from '../ConfigService'
|
||||
import { getCustomInstructionService } from '../CustomInstructionService'
|
||||
import { isUuid, type PushContext } from './sync-adapters'
|
||||
import { toSyncRemoteError, type RemoteRow, type SyncRemoteError } from './sync-types'
|
||||
|
||||
export const SETTINGS_ROW_ID = 'self'
|
||||
|
||||
/** 이 키들이 바뀌면 설정 동기화 대상이다 */
|
||||
export const SYNCED_CONFIG_KEYS = ['language', 'theme', 'defaultLLMAction', 'activeInstructionId'] as const
|
||||
|
||||
const SETTINGS_COLUMNS = 'user_id,locale,theme_mode,auto_polish_enabled,active_instruction_id,revision'
|
||||
|
||||
let applyingRemote = false
|
||||
|
||||
/** 원격 설정을 반영하는 중인지 — 이 동안의 설정 변경은 다시 올리지 않는다(에코 방지) */
|
||||
export function isApplyingRemoteSettings(): boolean {
|
||||
return applyingRemote
|
||||
}
|
||||
|
||||
export interface LocalSettings {
|
||||
language: string
|
||||
theme: ThemeMode
|
||||
defaultLLMAction: LLMActionSelection
|
||||
activeInstructionId: string
|
||||
}
|
||||
|
||||
export interface RemoteSettingsPatch {
|
||||
locale?: 'ko' | 'en'
|
||||
theme_mode?: 'system' | 'light' | 'dark'
|
||||
auto_polish_enabled: boolean
|
||||
}
|
||||
|
||||
export function readLocalSettings(): LocalSettings {
|
||||
return {
|
||||
language: String(configGet('language') ?? 'ko'),
|
||||
theme: (configGet('theme') ?? 'auto') as ThemeMode,
|
||||
defaultLLMAction: (configGet('defaultLLMAction') ?? 'refine') as LLMActionSelection,
|
||||
activeInstructionId: String(configGet('activeInstructionId') ?? ''),
|
||||
}
|
||||
}
|
||||
|
||||
export function toRemoteSettings(local: LocalSettings): RemoteSettingsPatch {
|
||||
const patch: RemoteSettingsPatch = { auto_polish_enabled: local.defaultLLMAction !== 'none' }
|
||||
if (local.language === 'ko' || local.language === 'en') patch.locale = local.language
|
||||
if (local.theme === 'auto') patch.theme_mode = 'system'
|
||||
else if (local.theme === 'light' || local.theme === 'dark') patch.theme_mode = local.theme
|
||||
return patch
|
||||
}
|
||||
|
||||
/** 원격 행이 요구하는 로컬 변경만 돌려준다(같으면 빈 객체). */
|
||||
export function remoteToLocalPatch(
|
||||
row: RemoteRow,
|
||||
local: LocalSettings,
|
||||
userInstructionIds: ReadonlySet<string>
|
||||
): Partial<LocalSettings> {
|
||||
const patch: Partial<LocalSettings> = {}
|
||||
if ((row.locale === 'ko' || row.locale === 'en') && row.locale !== local.language) patch.language = row.locale
|
||||
|
||||
const sharedTheme = local.theme === 'auto' || local.theme === 'light' || local.theme === 'dark'
|
||||
if (sharedTheme) {
|
||||
const theme: ThemeMode | null =
|
||||
row.theme_mode === 'system' ? 'auto' : row.theme_mode === 'light' || row.theme_mode === 'dark' ? row.theme_mode : null
|
||||
if (theme && theme !== local.theme) patch.theme = theme
|
||||
}
|
||||
|
||||
if (row.auto_polish_enabled === false && local.defaultLLMAction !== 'none') patch.defaultLLMAction = 'none'
|
||||
if (row.auto_polish_enabled === true && local.defaultLLMAction === 'none') patch.defaultLLMAction = 'refine'
|
||||
|
||||
const remoteActive = row.active_instruction_id
|
||||
if (isUuid(remoteActive)) {
|
||||
if (userInstructionIds.has(remoteActive) && remoteActive !== local.activeInstructionId) {
|
||||
patch.activeInstructionId = remoteActive
|
||||
}
|
||||
} else if (remoteActive === null && userInstructionIds.has(local.activeInstructionId)) {
|
||||
// 사용자 명령이 해제됐다. 데스크톱 프리셋(builtin-*)이 켜져 있으면 모바일이 알 수 없으므로 둔다.
|
||||
patch.activeInstructionId = ''
|
||||
}
|
||||
return patch
|
||||
}
|
||||
|
||||
function userInstructionIds(): Set<string> {
|
||||
return new Set(
|
||||
getCustomInstructionService()
|
||||
.getAll()
|
||||
.filter((i) => !i.isBuiltin && isUuid(i.id))
|
||||
.map((i) => i.id)
|
||||
)
|
||||
}
|
||||
|
||||
export async function fetchRemoteSettings(ctx: PushContext): Promise<RemoteRow | null> {
|
||||
const rows = await ctx.remote.selectWhere('user_settings', ctx.userId, [], SETTINGS_COLUMNS)
|
||||
return rows[0] ?? null
|
||||
}
|
||||
|
||||
/** 로컬 설정을 올린다. 모바일과 같은 revision 규칙(재조회 → revision 조건부 수정, 최대 3회). */
|
||||
export async function pushSettings(ctx: PushContext): Promise<SyncRemoteError | null> {
|
||||
const local = readLocalSettings()
|
||||
const desired = toRemoteSettings(local)
|
||||
try {
|
||||
let remote: RemoteRow | null = null
|
||||
let saved = false
|
||||
for (let attempt = 0; attempt < 3 && !saved; attempt++) {
|
||||
remote = await fetchRemoteSettings(ctx)
|
||||
if (!remote) {
|
||||
try {
|
||||
await ctx.remote.insert('user_settings', [{ user_id: ctx.userId, ...desired }])
|
||||
saved = true
|
||||
} catch (err) {
|
||||
const error = toSyncRemoteError(err)
|
||||
if (error.code !== '23505') throw error
|
||||
}
|
||||
continue
|
||||
}
|
||||
const revision = Number(remote.revision ?? 1)
|
||||
// 이미 같으면 쓰지 않는다 — revision만 올리면 다른 기기에 가짜 변경이 전파된다.
|
||||
const current = remote
|
||||
if ((Object.keys(desired) as Array<keyof RemoteSettingsPatch>).every((k) => current[k] === desired[k])) {
|
||||
saved = true
|
||||
break
|
||||
}
|
||||
const changed = await ctx.remote.updateMatching(
|
||||
'user_settings',
|
||||
ctx.userId,
|
||||
{ revision },
|
||||
{ ...desired, revision: revision + 1 }
|
||||
)
|
||||
saved = changed > 0
|
||||
}
|
||||
if (!saved) return toSyncRemoteError({ code: '40001', message: 'user_settings changed repeatedly on another device' })
|
||||
|
||||
// 활성 명령: 양쪽에 있는 사용자 명령이거나 '없음'일 때만 올린다(프리셋은 모바일에 없다).
|
||||
const ids = userInstructionIds()
|
||||
const active = local.activeInstructionId
|
||||
const remoteActive = remote?.active_instruction_id ?? null
|
||||
if (active === '' && remoteActive !== null) {
|
||||
await ctx.remote.rpc('set_active_custom_instruction', { instruction_id: null })
|
||||
} else if (ids.has(active) && remoteActive !== active) {
|
||||
await ctx.remote.rpc('set_active_custom_instruction', { instruction_id: active })
|
||||
}
|
||||
return null
|
||||
} catch (err) {
|
||||
return toSyncRemoteError(err)
|
||||
}
|
||||
}
|
||||
|
||||
/** 원격 설정을 로컬에 반영한다. 바꾼 것이 있으면 true */
|
||||
export function applyRemoteSettings(row: RemoteRow): boolean {
|
||||
const patch = remoteToLocalPatch(row, readLocalSettings(), userInstructionIds())
|
||||
const keys = Object.keys(patch) as Array<keyof LocalSettings>
|
||||
if (keys.length === 0) return false
|
||||
applyingRemote = true
|
||||
try {
|
||||
if (patch.language !== undefined) configSet('language', patch.language)
|
||||
if (patch.theme !== undefined) configSet('theme', patch.theme)
|
||||
if (patch.defaultLLMAction !== undefined) configSet('defaultLLMAction', patch.defaultLLMAction)
|
||||
if (patch.activeInstructionId !== undefined) configSet('activeInstructionId', patch.activeInstructionId)
|
||||
} finally {
|
||||
applyingRemote = false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
|
@ -76,9 +76,66 @@ export class SupabaseSyncRemote implements SyncRemote {
|
|||
return rowsOf(data)
|
||||
}
|
||||
|
||||
async upsert(table: string, rows: RemoteRow[]): Promise<void> {
|
||||
async selectChildren(
|
||||
table: string,
|
||||
parentColumn: string,
|
||||
parentId: string,
|
||||
columns: string,
|
||||
orderBy: string
|
||||
): Promise<RemoteRow[]> {
|
||||
const { data, error } = await this.client
|
||||
.from(table)
|
||||
.select(columns)
|
||||
.eq(parentColumn, parentId)
|
||||
.order(orderBy, { ascending: true })
|
||||
if (error) throw toSyncRemoteError(error)
|
||||
return rowsOf(data)
|
||||
}
|
||||
|
||||
async upsert(table: string, rows: RemoteRow[], onConflict = 'id'): Promise<void> {
|
||||
if (rows.length === 0) return
|
||||
const { error } = await this.client.from(table).upsert(rows, { onConflict: 'id' })
|
||||
const { error } = await this.client.from(table).upsert(rows, { onConflict })
|
||||
if (error) throw toSyncRemoteError(error)
|
||||
}
|
||||
|
||||
async insert(table: string, rows: RemoteRow[]): Promise<void> {
|
||||
if (rows.length === 0) return
|
||||
const { error } = await this.client.from(table).insert(rows)
|
||||
if (error) throw toSyncRemoteError(error)
|
||||
}
|
||||
|
||||
async updateMatching(
|
||||
table: string,
|
||||
userId: string,
|
||||
match: Record<string, string | number>,
|
||||
patch: RemoteRow
|
||||
): Promise<number> {
|
||||
let query = this.client.from(table).update(patch).eq('user_id', userId)
|
||||
for (const [column, value] of Object.entries(match)) query = query.eq(column, value)
|
||||
const { data, error } = await query.select('user_id')
|
||||
if (error) throw toSyncRemoteError(error)
|
||||
return rowsOf(data).length
|
||||
}
|
||||
|
||||
async deleteChildren(table: string, parentColumn: string, parentId: string): Promise<void> {
|
||||
const { error } = await this.client.from(table).delete().eq(parentColumn, parentId)
|
||||
if (error) throw toSyncRemoteError(error)
|
||||
}
|
||||
|
||||
async invokeFunction(name: string, body: Record<string, unknown>): Promise<unknown> {
|
||||
const { data, error } = await this.client.functions.invoke(name, { body })
|
||||
if (error) throw toSyncRemoteError(error)
|
||||
return data
|
||||
}
|
||||
|
||||
async uploadObject(bucket: string, key: string, bytes: Uint8Array, contentType: string): Promise<void> {
|
||||
const { error } = await this.client.storage.from(bucket).upload(key, bytes, { contentType, upsert: true })
|
||||
if (error) throw toSyncRemoteError(error)
|
||||
}
|
||||
|
||||
async removeObjects(bucket: string, keys: string[]): Promise<void> {
|
||||
if (keys.length === 0) return
|
||||
const { error } = await this.client.storage.from(bucket).remove(keys)
|
||||
if (error) throw toSyncRemoteError(error)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -10,11 +10,14 @@ import {
|
|||
meetingMemos,
|
||||
meetingSessions,
|
||||
memoTags,
|
||||
ragDocuments,
|
||||
} from '../../db/schema'
|
||||
import type { CustomInstruction, DictationTemplate, MeetingDocTemplate, TemplateField } from '@d3ro/core/types'
|
||||
import { getCustomInstructionService } from '../CustomInstructionService'
|
||||
import { getDictationTemplateService } from '../DictationTemplateService'
|
||||
import { getMeetingDocTemplateService } from '../MeetingDocTemplateService'
|
||||
import { getRAGService } from '../RAGService'
|
||||
import { purgeRemoteAudio } from './audio-sync'
|
||||
import { dropEntry } from './sync-outbox'
|
||||
import {
|
||||
SyncRemoteError,
|
||||
|
|
@ -43,8 +46,8 @@ export interface SyncAdapter {
|
|||
listLocalVersions(): LocalVersion[]
|
||||
push(ctx: PushContext, ids: string[]): Promise<PushOutcome[]>
|
||||
pushDeletes(ctx: PushContext, ids: string[]): Promise<PushOutcome[]>
|
||||
/** 원격 행을 로컬에 반영. 반영했으면 true */
|
||||
applyRemote(row: RemoteRow): boolean
|
||||
/** 원격 행을 로컬에 반영. 반영했으면 true. 자식 행을 더 읽어야 하는 엔티티는 비동기다 */
|
||||
applyRemote(row: RemoteRow, ctx: PushContext): boolean | Promise<boolean>
|
||||
/** 원격 삭제를 로컬에 반영. 지운 행이 있으면 true */
|
||||
deleteLocal(id: string): boolean
|
||||
}
|
||||
|
|
@ -209,7 +212,9 @@ const historyAdapter: SyncAdapter = {
|
|||
)
|
||||
return [...outcomes, ...missingAsDone(ids, found)]
|
||||
},
|
||||
pushDeletes(ctx, ids) {
|
||||
async pushDeletes(ctx, ids) {
|
||||
// 지운 기록의 녹음 파일도 저장소에서 지운다(행은 SET NULL로 남아 고아가 된다).
|
||||
await purgeRemoteAudio(ctx, 'history_id', ids)
|
||||
return deleteRemote(ctx, 'history', ids)
|
||||
},
|
||||
applyRemote(row) {
|
||||
|
|
@ -296,11 +301,9 @@ async function adoptRemoteDictionaryRow(
|
|||
const match = matches.find((m) => isUuid(m.id) && m.id !== localId)
|
||||
if (!match || !isUuid(match.id)) return error
|
||||
const remoteId = match.id
|
||||
db.transaction((tx) => {
|
||||
tx.delete(dictionary).where(eq(dictionary.id, localId)).run()
|
||||
})
|
||||
db.delete(dictionary).where(eq(dictionary.id, localId)).run()
|
||||
dropEntry('dictionary', localId)
|
||||
dictionaryAdapter.applyRemote(match)
|
||||
void dictionaryAdapter.applyRemote(match, ctx)
|
||||
const merged = db.select().from(dictionary).where(eq(dictionary.id, remoteId)).get()
|
||||
if (merged && local.usageCount > merged.usageCount) {
|
||||
db.update(dictionary)
|
||||
|
|
@ -440,7 +443,8 @@ const meetingsAdapter: SyncAdapter = {
|
|||
)
|
||||
return [...outcomes, ...missingAsDone(ids, found)]
|
||||
},
|
||||
pushDeletes(ctx, ids) {
|
||||
async pushDeletes(ctx, ids) {
|
||||
await purgeRemoteAudio(ctx, 'meeting_id', ids)
|
||||
return deleteRemote(ctx, 'meetings', ids)
|
||||
},
|
||||
applyRemote(row) {
|
||||
|
|
@ -841,6 +845,95 @@ const userTemplatesAdapter: SyncAdapter = {
|
|||
},
|
||||
}
|
||||
|
||||
// ── knowledge_documents ─────────────────────────────────
|
||||
// 원문 청크만 옮긴다. 임베딩은 기기마다 모델·차원이 달라(데스크톱 nomic 768 vs 서버 1536) 각자 만든다.
|
||||
|
||||
const KNOWLEDGE_TYPES = ['txt', 'md', 'pdf', 'docx'] as const
|
||||
const CHUNK_BATCH = 200
|
||||
|
||||
async function pushKnowledgeDocument(ctx: PushContext, id: string): Promise<SyncRemoteError | null> {
|
||||
const db = getDatabase()
|
||||
const doc = db.select().from(ragDocuments).where(eq(ragDocuments.id, id)).get()
|
||||
if (!doc) return null
|
||||
const chunks = getRAGService().getStoredChunks(id)
|
||||
if (chunks.length === 0) return null
|
||||
try {
|
||||
await ctx.remote.upsert('knowledge_documents', [
|
||||
{
|
||||
id: doc.id,
|
||||
user_id: ctx.userId,
|
||||
title: doc.fileName.slice(0, 300),
|
||||
file_name: doc.fileName,
|
||||
file_type: doc.fileType,
|
||||
chunk_count: chunks.length,
|
||||
indexed: false,
|
||||
indexed_at: null,
|
||||
created_at: iso(doc.addedAt),
|
||||
},
|
||||
])
|
||||
await ctx.remote.deleteChildren('knowledge_chunks', 'document_id', doc.id)
|
||||
for (let i = 0; i < chunks.length; i += CHUNK_BATCH) {
|
||||
await ctx.remote.insert(
|
||||
'knowledge_chunks',
|
||||
chunks.slice(i, i + CHUNK_BATCH).map((content, offset) => ({
|
||||
document_id: doc.id,
|
||||
chunk_index: i + offset,
|
||||
content,
|
||||
}))
|
||||
)
|
||||
}
|
||||
} catch (err) {
|
||||
return toSyncRemoteError(err)
|
||||
}
|
||||
// 모바일·웹 검색용 서버 임베딩. 공급자 키가 없거나 한도에 걸려도 원문은 이미 올라갔다 —
|
||||
// 모바일에서 "색인" 을 다시 누를 수 있으므로 여기서 실패로 되돌리지 않는다.
|
||||
try {
|
||||
await ctx.remote.invokeFunction('embed-chunks', { document_id: doc.id })
|
||||
} catch {
|
||||
// 의도적 무시(위 주석)
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
const knowledgeAdapter: SyncAdapter = {
|
||||
entity: 'knowledge_documents',
|
||||
pull: { columns: 'id,title,file_name,file_type,chunk_count,created_at,updated_at' },
|
||||
listLocalVersions() {
|
||||
return getDatabase()
|
||||
.select({ id: ragDocuments.id, addedAt: ragDocuments.addedAt })
|
||||
.from(ragDocuments)
|
||||
.all()
|
||||
.filter((r) => isUuid(r.id))
|
||||
.map((r) => ({ id: r.id, updatedAt: r.addedAt }))
|
||||
},
|
||||
async push(ctx, ids) {
|
||||
const outcomes: PushOutcome[] = []
|
||||
for (const id of ids) outcomes.push({ id, error: await pushKnowledgeDocument(ctx, id) })
|
||||
return outcomes
|
||||
},
|
||||
pushDeletes(ctx, ids) {
|
||||
return deleteRemote(ctx, 'knowledge_documents', ids)
|
||||
},
|
||||
async applyRemote(row, ctx) {
|
||||
const id = row.id
|
||||
if (!isUuid(id)) return false
|
||||
const exists = getDatabase().select({ id: ragDocuments.id }).from(ragDocuments).where(eq(ragDocuments.id, id)).get()
|
||||
if (exists) return false
|
||||
const chunkRows = await ctx.remote.selectChildren('knowledge_chunks', 'document_id', id, 'chunk_index,content', 'chunk_index')
|
||||
const chunks = chunkRows.map((c) => str(c.content)).filter((c): c is string => c !== null)
|
||||
return getRAGService().applyRemoteDocument({
|
||||
id,
|
||||
fileName: str(row.file_name) ?? str(row.title) ?? 'document',
|
||||
fileType: oneOf(row.file_type, KNOWLEDGE_TYPES) ?? 'txt',
|
||||
chunks,
|
||||
addedAt: ms(row.created_at, Date.now()),
|
||||
})
|
||||
},
|
||||
deleteLocal(id) {
|
||||
return getRAGService().removeRemote(id)
|
||||
},
|
||||
}
|
||||
|
||||
/** push 순서: 부모(이력·회의)가 자식(태그·메모·문서)보다 먼저. 삭제는 역순. */
|
||||
export const TABLE_ADAPTERS: readonly SyncAdapter[] = [
|
||||
historyAdapter,
|
||||
|
|
@ -850,6 +943,7 @@ export const TABLE_ADAPTERS: readonly SyncAdapter[] = [
|
|||
meetingDocumentsAdapter,
|
||||
customInstructionsAdapter,
|
||||
userTemplatesAdapter,
|
||||
knowledgeAdapter,
|
||||
]
|
||||
|
||||
export function adapterFor(entity: SyncEntity): SyncAdapter | null {
|
||||
|
|
|
|||
|
|
@ -11,6 +11,12 @@ export const SYNC_ENTITIES = [
|
|||
'memo_tags',
|
||||
'custom_instructions',
|
||||
'user_templates',
|
||||
'knowledge_documents',
|
||||
/** 단일 행(row_id 'self'): 언어·테마·자동 다듬기·활성 명령 */
|
||||
'user_settings',
|
||||
/** 녹음 파일 업로드(row_id = history id / meeting id) */
|
||||
'history_audio',
|
||||
'meeting_audio',
|
||||
] as const
|
||||
|
||||
export type SyncEntity = (typeof SYNC_ENTITIES)[number]
|
||||
|
|
@ -54,9 +60,18 @@ export interface RemotePageRequest {
|
|||
export interface SyncRemote {
|
||||
fetchPage(request: RemotePageRequest): Promise<RemoteRow[]>
|
||||
selectWhere(table: string, userId: string, filters: RemoteFilter[], columns?: string): Promise<RemoteRow[]>
|
||||
upsert(table: string, rows: RemoteRow[]): Promise<void>
|
||||
/** user_id 소유 검사가 없는 자식 테이블(knowledge_chunks) 조회 — RLS가 부모 소유로 거른다 */
|
||||
selectChildren(table: string, parentColumn: string, parentId: string, columns: string, orderBy: string): Promise<RemoteRow[]>
|
||||
upsert(table: string, rows: RemoteRow[], onConflict?: string): Promise<void>
|
||||
insert(table: string, rows: RemoteRow[]): Promise<void>
|
||||
/** user_id + match 조건에 맞는 행을 patch로 수정하고 수정된 행 수를 돌려준다(낙관적 동시성용) */
|
||||
updateMatching(table: string, userId: string, match: Record<string, string | number>, patch: RemoteRow): Promise<number>
|
||||
deleteByIds(table: string, userId: string, ids: string[]): Promise<void>
|
||||
deleteChildren(table: string, parentColumn: string, parentId: string): Promise<void>
|
||||
rpc(name: string, params: Record<string, unknown>): Promise<unknown>
|
||||
invokeFunction(name: string, body: Record<string, unknown>): Promise<unknown>
|
||||
uploadObject(bucket: string, key: string, bytes: Uint8Array, contentType: string): Promise<void>
|
||||
removeObjects(bucket: string, keys: string[]): Promise<void>
|
||||
}
|
||||
|
||||
export class SyncRemoteError extends Error {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue