From efe7cfdb3321b5df4921b6dc6727b095b522e9dd Mon Sep 17 00:00:00 2001 From: Askhz <1361267452@qq.com> Date: Thu, 21 May 2026 19:31:17 +0800 Subject: [PATCH] feat: add memory cloud sync to costrict-web API Auto-memory files are now synced to the cloud via /api/memories endpoint. On startup, a full scan uploads new/changed files. A file watcher handles incremental changes with 10s debounce. A process lock ensures only one watcher per project. Disabled via DISABLE_MEMORY_CLOUD_SYNC=1. Co-Authored-By: CoStrict-DeepSeek-V4-Pro --- .../services/memoryCloudSync/index.ts | 283 ++++++++++++++++ .../services/memoryCloudSync/types.ts | 77 +++++ .../services/memoryCloudSync/watcher.ts | 313 ++++++++++++++++++ src/setup.ts | 3 + 4 files changed, 676 insertions(+) create mode 100644 src/costrict/services/memoryCloudSync/index.ts create mode 100644 src/costrict/services/memoryCloudSync/types.ts create mode 100644 src/costrict/services/memoryCloudSync/watcher.ts diff --git a/src/costrict/services/memoryCloudSync/index.ts b/src/costrict/services/memoryCloudSync/index.ts new file mode 100644 index 000000000..49c5bedfa --- /dev/null +++ b/src/costrict/services/memoryCloudSync/index.ts @@ -0,0 +1,283 @@ +/** + * Memory Cloud Sync Service + * + * Syncs auto-memory files (personal memory) between the local filesystem + * and the costrict-web cloud API. + * + * API contract (costrict-web server/internal/memory): + * POST /api/memories → create a new memory + * PUT /api/memories/:id → update an existing memory + * + * Sync semantics: + * - Initial scan: on startup, scan all .md files, compare against local + * .cloud_sync_state.json, upload new/changed files. + * - Incremental: file watcher triggers re-scan of changed files. + * - Content MD5 comparison prevents redundant uploads. + * - Deletions do NOT propagate to the server. + */ + +import { createHash } from 'crypto' +import { readdir, readFile, writeFile } from 'fs/promises' +import { basename, join } from 'path' +import { getAutoMemPath, isAutoMemoryEnabled } from '../../../memdir/paths.js' +import { + getCoStrictBaseURL, +} from '../../../costrict/provider/auth.js' +import { loadCoStrictCredentials } from '../../../costrict/provider/credentials.js' +import { parseFrontmatter } from '../../../utils/frontmatterParser.js' +import { getProjectRoot } from '../../../bootstrap/state.js' +import type { + CloudSyncState, + CreateMemoryRequest, + MemoryResponse, + MemoryUploadResult, + SyncedFileState, + UpdateMemoryRequest, +} from './types.js' + +const API_TIMEOUT_MS = 30_000 + +// ─── Auth & endpoint ────────────────────────────────────────── + +function getMemoryAPIEndpoint(path: string): string { + const baseUrl = getCoStrictBaseURL() + return `${baseUrl}/cloud-api/api/memories${path}` +} + +async function getAuthHeaders(): Promise<{ + headers?: Record + error?: string +}> { + const creds = await loadCoStrictCredentials() + if (creds?.access_token) { + return { + headers: { + Authorization: `Bearer ${creds.access_token}`, + 'Content-Type': 'application/json', + }, + } + } + return { error: 'No CoStrict credentials available' } +} + +/** + * Check if cloud sync can proceed (feature gating is done in watcher). + */ +export function isMemoryCloudSyncAvailable(): boolean { + return isAutoMemoryEnabled() +} + +// ─── State file ──────────────────────────────────────────────── + +const STATE_FILENAME = '.cloud_sync_state.json' + +function getStateFilePath(): string { + return join(getAutoMemPath(), STATE_FILENAME) +} + +export async function loadSyncState(): Promise { + try { + const raw = await readFile(getStateFilePath(), 'utf-8') + const parsed = JSON.parse(raw) + return { items: parsed.items ?? {} } + } catch { + return { items: {} } + } +} + +export async function saveSyncState(state: CloudSyncState): Promise { + await writeFile( + getStateFilePath(), + JSON.stringify(state, null, 2) + '\n', + 'utf-8', + ) +} + +// ─── Content hash ────────────────────────────────────────────── + +export function computeMD5(content: string): string { + return createHash('md5').update(content, 'utf8').digest('hex') +} + +// ─── Upload ──────────────────────────────────────────────────── + +/** + * Upload a single memory file to the cloud. + * Uses POST for new memories, PUT for existing ones. + */ +export async function uploadMemory( + slug: string, + content: string, + existingState?: SyncedFileState, +): Promise { + try { + const auth = await getAuthHeaders() + if (auth.error) { + return { + success: false, + slug, + error: auth.error, + errorType: 'auth', + } + } + + const { frontmatter } = parseFrontmatter(content) + const name = (frontmatter.name as string) || slug + const description = (frontmatter.description as string) || '' + const type = (frontmatter.type as string) || 'user' + const projectPath = getProjectRoot() || process.cwd() + const workDir = process.cwd() + + if (existingState?.memoryId) { + // Existing memory → PUT to update + const body: UpdateMemoryRequest = { + name, + description, + content, + bumpVersion: true, + } + + const url = getMemoryAPIEndpoint(`/${existingState.memoryId}`) + const response = await fetch(url, { + method: 'PUT', + headers: auth.headers, + body: JSON.stringify(body), + signal: AbortSignal.timeout(API_TIMEOUT_MS), + }) + + if (!response.ok) { + return { + success: false, + slug, + error: `PUT failed: ${response.status} ${response.statusText}`, + errorType: response.status >= 500 ? 'network' : 'unknown', + httpStatus: response.status, + } + } + + const data = (await response.json()) as MemoryResponse + return { + success: true, + slug, + memoryId: data.id, + version: data.currentVersion, + } + } + + // New memory → POST to create + const body: CreateMemoryRequest = { + name, + slug, + projectPath, + workDir, + type, + description, + content, + } + + const url = getMemoryAPIEndpoint('') + const response = await fetch(url, { + method: 'POST', + headers: auth.headers, + body: JSON.stringify(body), + signal: AbortSignal.timeout(API_TIMEOUT_MS), + }) + + if (!response.ok) { + return { + success: false, + slug, + error: `POST failed: ${response.status} ${response.statusText}`, + errorType: response.status >= 500 ? 'network' : 'unknown', + httpStatus: response.status, + } + } + + const data = (await response.json()) as MemoryResponse + return { + success: true, + slug, + memoryId: data.id, + version: data.currentVersion, + } + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + const isTimeout = + message.includes('timeout') || message.includes('aborted') + return { + success: false, + slug, + error: message, + errorType: isTimeout ? 'timeout' : 'network', + } + } +} + +// ─── Full scan & sync ────────────────────────────────────────── + +/** + * Scan all .md files in the auto-memory directory and sync them to the cloud. + * Skips MEMORY.md and files in logs/ directory. + * Returns the set of slugs that were successfully synced. + */ +export async function scanAndSync( + existingState: CloudSyncState, +): Promise<{ results: MemoryUploadResult[]; newState: CloudSyncState }> { + const memoryDir = getAutoMemPath() + const newState: CloudSyncState = { items: { ...existingState.items } } + const results: MemoryUploadResult[] = [] + + let entries: string[] + try { + entries = await readdir(memoryDir, { recursive: true }) + } catch { + // Memory dir doesn't exist yet — nothing to sync + return { results, newState } + } + + const mdFiles = entries.filter( + f => + f.endsWith('.md') && + basename(f) !== 'MEMORY.md' && + !f.startsWith('logs/') && + !f.startsWith('logs\\'), + ) + + // Process files sequentially to avoid overwhelming the API + for (const relativePath of mdFiles) { + const slug = basename(relativePath, '.md') + const filePath = join(memoryDir, relativePath) + + try { + const raw = await readFile(filePath, 'utf-8') + const md5 = computeMD5(raw) + const existing = newState.items[slug] + + // Skip if content hasn't changed + if (existing && existing.contentMD5 === md5) { + continue + } + + const result = await uploadMemory(slug, raw, existing) + + if (result.success && result.memoryId) { + newState.items[slug] = { + memoryId: result.memoryId, + contentMD5: md5, + version: result.version ?? (existing?.version ?? 0) + 1, + } + } + + results.push(result) + } catch (err) { + results.push({ + success: false, + slug, + error: err instanceof Error ? err.message : String(err), + errorType: 'unknown', + }) + } + } + + return { results, newState } +} \ No newline at end of file diff --git a/src/costrict/services/memoryCloudSync/types.ts b/src/costrict/services/memoryCloudSync/types.ts new file mode 100644 index 000000000..17ff5ff15 --- /dev/null +++ b/src/costrict/services/memoryCloudSync/types.ts @@ -0,0 +1,77 @@ +/** + * Memory Cloud Sync Types + * + * Type definitions for the memory cloud sync API, which uploads + * auto-memory files to the costrict-web /api/memories endpoint. + */ + +/** + * POST /api/memories request body + */ +export type CreateMemoryRequest = { + name: string + slug: string + projectPath: string + workDir: string + type: string + description: string + content: string +} + +/** + * PUT /api/memories/:id request body + */ +export type UpdateMemoryRequest = { + name?: string + description?: string + content?: string + bumpVersion: boolean +} + +/** + * Response from creating or fetching a memory + */ +export type MemoryResponse = { + id: string + userId: string + projectPath: string + slug: string + name: string + type: string + description?: string + currentVersion: number + createdAt: string + updatedAt: string +} + +/** + * Per-file sync state stored in .cloud_sync_state.json + */ +export type SyncedFileState = { + /** Server-side memory ID */ + memoryId: string + /** MD5 hash of the file content at last sync */ + contentMD5: string + /** Server-side version number */ + version: number +} + +/** + * Full sync state file content + */ +export type CloudSyncState = { + items: Record +} + +/** + * Result of uploading a single memory file + */ +export type MemoryUploadResult = { + success: boolean + slug: string + memoryId?: string + version?: number + error?: string + errorType?: 'auth' | 'network' | 'timeout' | 'unknown' + httpStatus?: number +} \ No newline at end of file diff --git a/src/costrict/services/memoryCloudSync/watcher.ts b/src/costrict/services/memoryCloudSync/watcher.ts new file mode 100644 index 000000000..37af48ecf --- /dev/null +++ b/src/costrict/services/memoryCloudSync/watcher.ts @@ -0,0 +1,313 @@ +/** + * Memory Cloud Sync File Watcher + * + * On startup: scans all existing .md files and syncs them to the cloud. + * After startup: watches the auto-memory directory for changes and triggers + * a debounced push to the server when files are modified. + */ + +import { type FSWatcher, watch } from 'fs' +import { mkdir, readFile, unlink, writeFile } from 'fs/promises' +import { join } from 'path' +import { getAutoMemPath, isAutoMemoryEnabled } from '../../../memdir/paths.js' +import { loadCoStrictCredentials } from '../../../costrict/provider/credentials.js' +import { registerCleanup } from '../../../utils/cleanupRegistry.js' +import { logForDebugging } from '../../../utils/debug.js' +import { isEnvTruthy } from '../../../utils/envUtils.js' +import { errorMessage } from '../../../utils/errors.js' +import { isProcessRunning } from '../../../utils/genericProcessUtils.js' +import { + loadSyncState, + saveSyncState, + scanAndSync, +} from './index.js' + +const DEBOUNCE_MS = 10_000 // Wait 10s after last change before pushing +const LOCK_FILE = '.cloud_sync.lock' +// PID recycling guard: if the lock is older than this, reclaim even if PID is alive +const LOCK_STALE_MS = 60 * 60 * 1000 // 1 hour + +// ─── Watcher state ───────────────────────────────────────────── +let watcher: FSWatcher | null = null +let debounceTimer: ReturnType | null = null +let pushInProgress = false +let hasPendingChanges = false +let watcherStarted = false +let lockAcquired = false + +// ─── Process lock ───────────────────────────────────────────── + +function lockPath(): string { + return join(getAutoMemPath(), LOCK_FILE) +} + +/** + * Try to acquire the cloud sync lock. Only one process per project should + * run the watcher — others skip to avoid duplicating uploads and corrupting + * the shared .cloud_sync_state.json. + */ +async function tryAcquireLock(): Promise { + const path = lockPath() + try { + const raw = await readFile(path, 'utf8') + const holderPid = parseInt(raw.trim(), 10) + + if (Number.isFinite(holderPid) && isProcessRunning(holderPid)) { + const stat = await import('fs/promises').then(m => m.stat(path)) + const age = Date.now() - stat.mtimeMs + if (age < LOCK_STALE_MS) { + logForDebugging( + `memory-cloud-sync: lock held by live PID ${holderPid}, skipping`, + { level: 'debug' }, + ) + return false + } + // Lock is stale — reclaim below + } + } catch { + // ENOENT — no prior lock, proceed + } + + await mkdir(getAutoMemPath(), { recursive: true }) + await writeFile(path, String(process.pid)) + + // Re-read to verify we won the race + const verify = await readFile(path, 'utf8') + if (parseInt(verify.trim(), 10) !== process.pid) { + logForDebugging( + 'memory-cloud-sync: lost lock race, skipping', + { level: 'debug' }, + ) + return false + } + + return true +} + +async function releaseLock(): Promise { + if (!lockAcquired) return + try { + const path = lockPath() + const raw = await readFile(path, 'utf8') + if (parseInt(raw.trim(), 10) === process.pid) { + await unlink(path) + } + } catch { + // Best-effort + } + lockAcquired = false +} + +/** + * Execute the push: re-scan all files and upload changed ones. + */ +async function executePush(): Promise { + pushInProgress = true + try { + const state = await loadSyncState() + const { results, newState } = await scanAndSync(state) + await saveSyncState(newState) + + const succeeded = results.filter(r => r.success).length + const failed = results.filter(r => !r.success).length + + if (succeeded > 0 || failed > 0) { + logForDebugging( + `memory-cloud-sync: pushed ${succeeded} files${ + failed > 0 ? `, ${failed} failed` : '' + }`, + { level: 'info' }, + ) + } + + if (failed === 0) { + hasPendingChanges = false + } + + // Log individual failures for debugging + for (const r of results) { + if (!r.success) { + logForDebugging( + `memory-cloud-sync: upload failed for "${r.slug}": ${r.error}`, + { level: 'warn' }, + ) + } + } + } catch (e) { + logForDebugging(`memory-cloud-sync: push error: ${errorMessage(e)}`, { + level: 'warn', + }) + } finally { + pushInProgress = false + } +} + +/** + * Debounced push: waits for writes to settle, then pushes once. + */ +function schedulePush(): void { + hasPendingChanges = true + if (debounceTimer) { + clearTimeout(debounceTimer) + } + debounceTimer = setTimeout(() => { + debounceTimer = null + if (pushInProgress) { + // A push is already running — mark pending and let it pick up + // the new changes on its next iteration + hasPendingChanges = true + return + } + void executePush() + }, DEBOUNCE_MS) +} + +/** + * Start watching the auto-memory directory for changes. + */ +async function startFileWatcher(memoryDir: string): Promise { + if (watcherStarted) { + return + } + watcherStarted = true + + try { + await mkdir(memoryDir, { recursive: true }) + + watcher = watch( + memoryDir, + { persistent: true, recursive: true }, + (_eventType, _filename) => { + schedulePush() + }, + ) + watcher.on('error', err => { + logForDebugging( + `memory-cloud-sync: fs.watch error: ${errorMessage(err)}`, + { level: 'warn' }, + ) + }) + logForDebugging(`memory-cloud-sync: watching ${memoryDir}`, { + level: 'debug', + }) + } catch (err) { + logForDebugging( + `memory-cloud-sync: failed to watch ${memoryDir}: ${errorMessage(err)}`, + { level: 'warn' }, + ) + } + + registerCleanup(async () => stopMemoryCloudWatcher()) +} + +/** + * Start the memory cloud sync system. + * + * Enabled by default. Set DISABLE_MEMORY_CLOUD_SYNC=1 to opt out. + * + * Returns early if: + * - DISABLE_MEMORY_CLOUD_SYNC env var is truthy + * - auto memory is disabled + * - user has no CoStrict credentials + * + * On start: performs a full scan of existing memory files and uploads + * any that are new or changed. Then starts the file watcher. + */ +export async function startMemoryCloudWatcher(): Promise { + if (isEnvTruthy(process.env.DISABLE_MEMORY_CLOUD_SYNC)) { + logForDebugging( + 'memory-cloud-sync: disabled via DISABLE_MEMORY_CLOUD_SYNC', + { level: 'debug' }, + ) + return + } + if (!isAutoMemoryEnabled()) { + logForDebugging( + 'memory-cloud-sync: auto memory disabled, skipping', + { level: 'debug' }, + ) + return + } + + const creds = await loadCoStrictCredentials() + if (!creds?.access_token) { + logForDebugging( + 'memory-cloud-sync: no CoStrict credentials, skipping', + { level: 'debug' }, + ) + return + } + + const memoryDir = getAutoMemPath() + + // Acquire process lock — only one watcher per project + const gotLock = await tryAcquireLock() + if (!gotLock) { + return + } + lockAcquired = true + + logForDebugging( + `memory-cloud-sync: starting initial scan of ${memoryDir}`, + { level: 'info' }, + ) + + // ── Initial full scan ────────────────────────────────────────── + try { + const state = await loadSyncState() + const { results, newState } = await scanAndSync(state) + await saveSyncState(newState) + + const succeeded = results.filter(r => r.success).length + const failed = results.filter(r => !r.success).length + + logForDebugging( + `memory-cloud-sync: initial scan complete — ${succeeded} uploaded${ + failed > 0 ? `, ${failed} failed` : '' + }`, + { level: 'info' }, + ) + } catch (err) { + logForDebugging( + `memory-cloud-sync: initial scan failed: ${errorMessage(err)}`, + { level: 'warn' }, + ) + } + + // ── Start file watcher ───────────────────────────────────────── + await startFileWatcher(memoryDir) +} + +/** + * Stop the file watcher and flush pending changes. + */ +export async function stopMemoryCloudWatcher(): Promise { + if (debounceTimer) { + clearTimeout(debounceTimer) + debounceTimer = null + } + if (watcher) { + watcher.close() + watcher = null + } + // Flush pending changes + if (hasPendingChanges) { + try { + await executePush() + } catch { + // Best-effort during shutdown + } + } + await releaseLock() +} + +/** + * Test-only: reset module state for test isolation. + */ +export function _resetWatcherStateForTesting(): void { + watcher = null + debounceTimer = null + pushInProgress = false + hasPendingChanges = false + watcherStarted = false +} \ No newline at end of file diff --git a/src/setup.ts b/src/setup.ts index 4befb9d5a..68e0c8522 100644 --- a/src/setup.ts +++ b/src/setup.ts @@ -363,6 +363,9 @@ export async function setup( m.startTeamMemoryWatcher(), ) // Start team memory sync watcher } + void import('./costrict/services/memoryCloudSync/watcher.js').then(m => + m.startMemoryCloudWatcher(), + ) // Start memory cloud sync watcher (disabled via DISABLE_MEMORY_CLOUD_SYNC=1) } initSinks() // Attach error log + analytics sinks and drain queued events