fix(sync): bind sync engine to its user DB, scope instructions/templates per account, harden navigation
This commit is contained in:
Yun Chan
parent
1b8fe445f3
commit
9aa7302944
30 changed files with 2614 additions and 386 deletions
|
|
@ -9,14 +9,14 @@ import {
|
|||
meetingDocuments,
|
||||
meetingMemos,
|
||||
meetingSessions,
|
||||
memoTags,
|
||||
ragDocuments,
|
||||
} from '../../db/schema'
|
||||
import type { CustomInstruction, DictationTemplate, MeetingDocTemplate, TemplateField } from '@d3ro/core/types'
|
||||
import { getCustomInstructionService } from '../CustomInstructionService'
|
||||
import { getCustomInstructionService, instructionLimitViolation } from '../CustomInstructionService'
|
||||
import { getDictationTemplateService } from '../DictationTemplateService'
|
||||
import { getMeetingDocTemplateService } from '../MeetingDocTemplateService'
|
||||
import { getRAGService } from '../RAGService'
|
||||
import { deleteHistoryLocally } from '../history-deletion'
|
||||
import { purgeRemoteAudio } from './audio-sync'
|
||||
import { pushMeetingSegments } from './transcript-sync'
|
||||
import { dropEntry } from './sync-outbox'
|
||||
|
|
@ -33,8 +33,31 @@ import {
|
|||
export interface PushContext {
|
||||
remote: SyncRemote
|
||||
userId: string
|
||||
/**
|
||||
* 이 작업이 아직 유효한지(엔진이 해제되지 않았고 같은 사용자 DB가 열려 있는지).
|
||||
* 네트워크를 기다린 뒤 로컬에 쓰는 어댑터는 쓰기 직전에 확인한다.
|
||||
*/
|
||||
isCurrent?: () => boolean
|
||||
}
|
||||
|
||||
/** 엔진이 해제됐거나 사용자 DB가 바뀌어 작업을 멈춘다 — 로컬에 아무것도 쓰지 않는다. */
|
||||
export class SyncAbortedError extends Error {
|
||||
constructor() {
|
||||
super('Sync aborted: engine disposed or user DB switched')
|
||||
this.name = 'SyncAbortedError'
|
||||
}
|
||||
}
|
||||
|
||||
export function assertCurrent(ctx: PushContext): void {
|
||||
if (ctx.isCurrent && !ctx.isCurrent()) throw new SyncAbortedError()
|
||||
}
|
||||
|
||||
/**
|
||||
* 원격 행 반영 결과. 'deferred' = 지금은 반영할 수 없지만(부모 회의가 아직 없음, 청크가 아직 없음)
|
||||
* 나중에 다시 시도해야 한다 — 엔진이 id를 기억했다가 다음 pull에서 다시 가져온다.
|
||||
*/
|
||||
export type ApplyResult = boolean | 'deferred'
|
||||
|
||||
export interface LocalVersion {
|
||||
id: string
|
||||
updatedAt: number
|
||||
|
|
@ -48,7 +71,7 @@ export interface SyncAdapter {
|
|||
push(ctx: PushContext, ids: string[]): Promise<PushOutcome[]>
|
||||
pushDeletes(ctx: PushContext, ids: string[]): Promise<PushOutcome[]>
|
||||
/** 원격 행을 로컬에 반영. 반영했으면 true. 자식 행을 더 읽어야 하는 엔티티는 비동기다 */
|
||||
applyRemote(row: RemoteRow, ctx: PushContext): boolean | Promise<boolean>
|
||||
applyRemote(row: RemoteRow, ctx: PushContext): ApplyResult | Promise<ApplyResult>
|
||||
/** 원격 삭제를 로컬에 반영. 지운 행이 있으면 true */
|
||||
deleteLocal(id: string): boolean
|
||||
}
|
||||
|
|
@ -192,7 +215,8 @@ export function historyToRemote(r: typeof history.$inferSelect, userId: string):
|
|||
summary_text: r.summaryText,
|
||||
is_favorite: r.isFavorite,
|
||||
created_at: iso(r.createdAt),
|
||||
updated_at: iso(r.updatedAt),
|
||||
// updated_at 은 보내지 않는다: 서버 시각(INSERT 기본값·트리거)이 keyset 커서의 기준이다.
|
||||
// 기기 시계·오프라인 편집 시각을 실으면 다른 기기의 커서 뒤로 들어가 영영 내려받히지 않는다.
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -252,9 +276,8 @@ const historyAdapter: SyncAdapter = {
|
|||
return true
|
||||
},
|
||||
deleteLocal(id) {
|
||||
const db = getDatabase()
|
||||
db.delete(memoTags).where(eq(memoTags.historyId, id)).run()
|
||||
return db.delete(history).where(eq(history.id, id)).run().changes > 0
|
||||
// 녹음 WAV까지 함께 지운다(폰에서 지운 기록의 목소리가 디스크에 남지 않게).
|
||||
return deleteHistoryLocally({ ids: [id] }).deleted > 0
|
||||
},
|
||||
}
|
||||
|
||||
|
|
@ -272,7 +295,6 @@ export function dictionaryToRemote(r: typeof dictionary.$inferSelect, userId: st
|
|||
usage_count: Math.round(r.usageCount),
|
||||
last_used_at: nullableIso(r.lastUsedAt),
|
||||
created_at: iso(r.createdAt),
|
||||
updated_at: iso(r.updatedAt),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -413,7 +435,6 @@ export function meetingToRemote(r: typeof meetingSessions.$inferSelect, userId:
|
|||
llm_latency_ms: int(r.llmLatencyMs),
|
||||
error_message: r.errorMessage,
|
||||
created_at: iso(r.createdAt),
|
||||
updated_at: iso(r.updatedAt),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -508,7 +529,6 @@ export function meetingMemoToRemote(r: typeof meetingMemos.$inferSelect, userId:
|
|||
content: r.content,
|
||||
timestamp_ms: Math.round(r.timestampMs),
|
||||
created_at: iso(r.createdAt),
|
||||
updated_at: iso(r.updatedAt ?? r.createdAt),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -540,8 +560,10 @@ const meetingMemosAdapter: SyncAdapter = {
|
|||
const id = row.id
|
||||
const meetingId = row.meeting_id
|
||||
const content = str(row.content)
|
||||
// 남의 팀 회의에 내가 단 메모는 부모 회의가 로컬에 없다 — 고아 행을 만들지 않는다.
|
||||
if (!isUuid(id) || !isUuid(meetingId) || content === null || !hasLocalMeeting(meetingId)) return false
|
||||
if (!isUuid(id) || !isUuid(meetingId) || content === null) return false
|
||||
// 부모 회의가 아직 로컬에 없으면 고아 행을 만들지 않고 미룬다(회의 pull이 실패했을 수 있다).
|
||||
// 남의 팀 회의에 단 내 메모는 끝내 부모가 오지 않는다 — 엔진이 몇 번 재시도 후 포기한다.
|
||||
if (!hasLocalMeeting(meetingId)) return 'deferred'
|
||||
const createdAt = ms(row.created_at, Date.now())
|
||||
const values = {
|
||||
sessionId: meetingId,
|
||||
|
|
@ -577,7 +599,6 @@ export function meetingDocumentToRemote(r: typeof meetingDocuments.$inferSelect,
|
|||
llm_model: r.llmModel,
|
||||
llm_latency_ms: int(r.llmLatencyMs),
|
||||
created_at: iso(r.createdAt),
|
||||
updated_at: iso(r.updatedAt),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -610,7 +631,7 @@ const meetingDocumentsAdapter: SyncAdapter = {
|
|||
const templateType = oneOf(row.template_type, DOCUMENT_TYPES)
|
||||
const title = str(row.title)
|
||||
if (!isUuid(id) || !isUuid(meetingId) || templateType === null || title === null) return false
|
||||
if (!hasLocalMeeting(meetingId)) return false
|
||||
if (!hasLocalMeeting(meetingId)) return 'deferred'
|
||||
const updatedAt = ms(row.updated_at, Date.now())
|
||||
const values = {
|
||||
sessionId: meetingId,
|
||||
|
|
@ -637,13 +658,24 @@ const meetingDocumentsAdapter: SyncAdapter = {
|
|||
// ── custom_instructions ─────────────────────────────────
|
||||
// 데스크톱 프리셋(builtin-*)과 서버 프리셋(builtin_key)은 각자 따로 있다. 사용자가 만든 명령만 맞춘다.
|
||||
|
||||
/**
|
||||
* 서버 CHECK(이름 80·설명 240·프롬프트 4000자)를 넘는 명령은 잘라 보내지 않는다 — 잘린 사본이 다음 pull에서
|
||||
* 원본을 덮는다. 재시도 불가 오류로 돌려 outbox가 보관(parked)하게 한다.
|
||||
*/
|
||||
export function instructionPushError(i: CustomInstruction): SyncRemoteError | null {
|
||||
const field = instructionLimitViolation(i)
|
||||
return field === null
|
||||
? null
|
||||
: new SyncRemoteError(`custom instruction ${field} exceeds the server length limit`, '22001', false)
|
||||
}
|
||||
|
||||
export function instructionToRemote(i: CustomInstruction, userId: string): RemoteRow {
|
||||
return {
|
||||
id: i.id,
|
||||
user_id: userId,
|
||||
name: i.name.trim().slice(0, 80),
|
||||
description: (i.description ?? '').trim().slice(0, 240),
|
||||
prompt: i.prompt.trim().slice(0, 4000),
|
||||
name: i.name.trim(),
|
||||
description: (i.description ?? '').trim(),
|
||||
prompt: i.prompt,
|
||||
icon: (i.icon || 'sparkles').slice(0, 32),
|
||||
sort_order: Math.max(0, Math.round(i.order)),
|
||||
created_at: iso(i.createdAt),
|
||||
|
|
@ -665,12 +697,19 @@ const customInstructionsAdapter: SyncAdapter = {
|
|||
.map((id) => service.getById(id))
|
||||
.filter((i): i is CustomInstruction => i !== null && !i.isBuiltin && isUuid(i.id))
|
||||
const found = new Set(items.map((i) => i.id))
|
||||
const tooLong: PushOutcome[] = []
|
||||
const sendable: CustomInstruction[] = []
|
||||
for (const item of items) {
|
||||
const error = instructionPushError(item)
|
||||
if (error) tooLong.push({ id: item.id, error })
|
||||
else sendable.push(item)
|
||||
}
|
||||
const outcomes = await upsertIsolated(
|
||||
ctx,
|
||||
'custom_instructions',
|
||||
items.map((i) => ({ id: i.id, row: instructionToRemote(i, ctx.userId) }))
|
||||
sendable.map((i) => ({ id: i.id, row: instructionToRemote(i, ctx.userId) }))
|
||||
)
|
||||
return [...outcomes, ...missingAsDone(ids, found)]
|
||||
return [...outcomes, ...tooLong, ...missingAsDone(ids, found)]
|
||||
},
|
||||
pushDeletes(ctx, ids) {
|
||||
return deleteRemote(ctx, 'custom_instructions', ids)
|
||||
|
|
@ -767,6 +806,15 @@ export function templateToRpcParams(local: LocalTemplate): Record<string, unknow
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* sync_upsert_user_template_v1 의 P0002(template_not_found)는 "이 id의 템플릿이 다른 계정 소유"라는 뜻이다.
|
||||
* 다시 보내도 같은 결과라 재시도 불가로 바꿔 outbox가 보관하게 한다(전역 P0002는 재시도 가능으로 남긴다).
|
||||
*/
|
||||
export function templateUpsertError(err: unknown): SyncRemoteError {
|
||||
const error = toSyncRemoteError(err)
|
||||
return error.code === 'P0002' ? new SyncRemoteError(error.message, error.code, false) : error
|
||||
}
|
||||
|
||||
const userTemplatesAdapter: SyncAdapter = {
|
||||
entity: 'user_templates',
|
||||
pull: { filters: [{ column: 'is_builtin', op: 'eq', value: false }] },
|
||||
|
|
@ -789,7 +837,7 @@ const userTemplatesAdapter: SyncAdapter = {
|
|||
await ctx.remote.rpc('sync_upsert_user_template_v1', templateToRpcParams(local))
|
||||
outcomes.push({ id, error: null })
|
||||
} catch (err) {
|
||||
outcomes.push({ id, error: toSyncRemoteError(err) })
|
||||
outcomes.push({ id, error: templateUpsertError(err) })
|
||||
}
|
||||
}
|
||||
return outcomes
|
||||
|
|
@ -932,7 +980,11 @@ const knowledgeAdapter: SyncAdapter = {
|
|||
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')
|
||||
// 네트워크를 기다리는 사이 로그아웃·계정 전환이 있었으면 다른 사용자 DB에 쓰지 않는다.
|
||||
assertCurrent(ctx)
|
||||
const chunks = chunkRows.map((c) => str(c.content)).filter((c): c is string => c !== null)
|
||||
// 다른 기기가 문서 행을 올리고 청크를 아직 못 올렸을 수 있다 — 다음 pull에서 다시 본다.
|
||||
if (chunks.length === 0) return 'deferred'
|
||||
return getRAGService().applyRemoteDocument({
|
||||
id,
|
||||
fileName: str(row.file_name) ?? str(row.title) ?? 'document',
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue