feat(sync): implement per-item sync with tombstone support

- Introduced per-item category data structure for granular sync of sessions and messages.
- Added tombstone management to track deleted items across machines.
- Implemented sharding for large item collections to optimize manifest size.
- Enhanced push and pull operations to handle item-level changes and tombstone propagation.
- Updated manifest structure to support sharded categories and tombstone metadata.
- Refactored packing and unpacking logic for items, including checksum calculations.
- Added helper functions for managing tombstones, including creation, expiration checks, and merging.
- Updated sync configuration to include tombstone grace period settings.

Author: ErcinDedeoglu

src/data/category-loader.ts

@@ -2,12 +2,20 @@
  * Category Loader
  *
  * Load data for sync categories.
+ * Supports both blob-based (config, state, etc.) and per-item (sessions, messages) loading.
  */
 
+import { readFile, readdir, stat } from 'node:fs/promises';
+import { join, basename } from 'node:path';
 import type { SyncCategory, PathConfig, SyncConfig } from '../types/index.js';
+import type { LocalSyncState } from '../types/sync.js';
+import type { Tombstone } from '../types/manifest.js';
+import { shouldUseItemSync } from '../types/manifest.js';
 import { getCategoryPaths } from '../types/paths.js';
-import type { CategoryData } from '../sync/operations/types.js';
+import type { CategoryData, BlobCategoryData, ItemCategoryData } from '../sync/operations/types.js';
 import { loadSinglePath, getPathKey } from './directory-loader.js';
+import { calculateChecksum } from '../sync/item-packer.js';
+import { createTombstone, detectLocalDeletions } from '../sync/tombstone.js';
 
 export interface LoadedData {
   categories: CategoryData[];
@@ -20,12 +28,22 @@ export interface LoadError {
   error: Error;
 }
 
+export interface LoadOptions {
+  /** Previous local state for deletion detection */
+  localState?: LocalSyncState | null;
+  /** Machine ID for tombstone creation */
+  machineId?: string;
+  /** Tombstone grace period in days (default: 30) */
+  tombstoneGraceDays?: number;
+}
+
 /**
  * Load all data for enabled categories.
  */
 export async function loadLocalData(
   pathConfig: PathConfig,
-  enabledCategories: SyncConfig['sync']
+  enabledCategories: SyncConfig['sync'],
+  options: LoadOptions = {}
 ): Promise<LoadedData> {
   const categoryPaths = getCategoryPaths(pathConfig);
   const categories: CategoryData[] = [];
@@ -35,7 +53,7 @@ export async function loadLocalData(
     if (!enabledCategories[category as SyncCategory]) continue;
 
     try {
-      const data = await loadCategoryData(category as SyncCategory, paths);
+      const data = await loadCategoryData(category as SyncCategory, paths, options);
       if (data) categories.push(data);
     } catch (error) {
       const firstPath = paths[0];
@@ -54,11 +72,26 @@ export async function loadLocalData(
 
 /**
  * Load data for a single category.
+ * Routes to blob or per-item loading based on category.
  */
 async function loadCategoryData(
   category: SyncCategory,
-  paths: string[]
+  paths: string[],
+  options: LoadOptions
 ): Promise<CategoryData | null> {
+  if (shouldUseItemSync(category)) {
+    return loadItemCategoryData(category, paths, options);
+  }
+  return loadBlobCategoryData(category, paths);
+}
+
+/**
+ * Load blob-based category data (legacy approach).
+ */
+async function loadBlobCategoryData(
+  category: SyncCategory,
+  paths: string[]
+): Promise<BlobCategoryData | null> {
   const categoryData: Record<string, unknown> = {};
   let hasData = false;
 
@@ -73,5 +106,97 @@ async function loadCategoryData(
   if (!hasData) return null;
 
   const isJsonl = category === 'state';
-  return { category, data: JSON.stringify(categoryData), isJsonl };
+  return { category, type: 'blob', data: JSON.stringify(categoryData), isJsonl };
+}
+
+/**
+ * Load per-item category data (sessions, messages).
+ * Each file is loaded as a separate item with its own checksum.
+ * Also detects locally deleted items and creates tombstones for them.
+ */
+async function loadItemCategoryData(
+  category: SyncCategory,
+  paths: string[],
+  options: LoadOptions
+): Promise<ItemCategoryData | null> {
+  const items: Record<string, string> = {};
+  const checksums: Record<string, string> = {};
+
+  for (const basePath of paths) {
+    await loadItemsFromPath(basePath, '', items, checksums);
+  }
+
+  // Detect locally deleted items by comparing with previous state
+  const tombstones: Record<string, Tombstone> = {};
+  const { localState, machineId, tombstoneGraceDays } = options;
+  const prevChecksums = localState?.itemChecksums;
+  const prevCategoryChecksums = prevChecksums ? prevChecksums[category] : undefined;
+  if (prevCategoryChecksums && machineId) {
+    const previousItemIds = new Set(Object.keys(prevCategoryChecksums));
+    const currentItemIds = new Set(Object.keys(items));
+    const deletedIds = detectLocalDeletions(currentItemIds, previousItemIds);
+
+    for (const itemId of deletedIds) {
+      tombstones[itemId] = createTombstone(itemId, machineId, tombstoneGraceDays);
+    }
+  }
+
+  // Return null only if no items AND no tombstones
+  if (Object.keys(items).length === 0 && Object.keys(tombstones).length === 0) {
+    return null;
+  }
+
+  const result: ItemCategoryData = { category, type: 'items', items, checksums };
+  if (Object.keys(tombstones).length > 0) {
+    result.tombstones = tombstones;
+  }
+  return result;
+}
+
+/**
+ * Recursively load items from a path.
+ * @param basePath - The base path being scanned
+ * @param relativePath - Current relative path from basePath
+ * @param items - Map to populate with item content
+ * @param checksums - Map to populate with item checksums
+ */
+async function loadItemsFromPath(
+  basePath: string,
+  relativePath: string,
+  items: Record<string, string>,
+  checksums: Record<string, string>
+): Promise<void> {
+  const fullPath = relativePath ? join(basePath, relativePath) : basePath;
+
+  try {
+    const stats = await stat(fullPath);
+
+    if (stats.isFile()) {
+      // Load file as an item
+      const content = await readFile(fullPath, 'utf-8');
+      const itemId = getItemId(basePath, relativePath);
+      items[itemId] = content;
+      checksums[itemId] = calculateChecksum(content);
+    } else if (stats.isDirectory()) {
+      // Recurse into directory
+      const entries = await readdir(fullPath, { withFileTypes: true });
+      for (const entry of entries) {
+        if (entry.name.startsWith('.')) continue; // Skip hidden files
+        const childRelPath = relativePath ? join(relativePath, entry.name) : entry.name;
+        await loadItemsFromPath(basePath, childRelPath, items, checksums);
+      }
+    }
+  } catch {
+    // Path doesn't exist - skip
+  }
+}
+
+/**
+ * Generate a unique item ID from base path and relative path.
+ * Example: basePath=/data/storage/session, relativePath=abc123.json → session/abc123.json
+ */
+function getItemId(basePath: string, relativePath: string): string {
+  const baseDir = basename(basePath);
+  if (!relativePath) return baseDir;
+  return `${baseDir}/${relativePath}`;
 }

src/data/index.ts

@@ -7,7 +7,7 @@
 export { loadLocalData } from './category-loader.js';
 export type { LoadedData, LoadError } from './category-loader.js';
 
-export { writeLocalData } from './writer.js';
+export { writeLocalData, deleteTombstonedItems } from './writer.js';
 
 export {
   loadConfig,

src/data/writer.ts

@@ -2,43 +2,103 @@
  * Data Writer
  *
  * Writes synced data back to local filesystem.
+ * Supports both blob-based overwrite and per-item merge writes.
  */
 
-import { mkdir, writeFile } from 'node:fs/promises';
-import { join } from 'node:path';
-import type { PathConfig } from '../types/index.js';
+import { mkdir, writeFile, unlink } from 'node:fs/promises';
+import { join, dirname } from 'node:path';
+import type { PathConfig, SyncCategory } from '../types/index.js';
 import { getCategoryPaths } from '../types/paths.js';
-import type { CategoryData } from '../sync/operations/types.js';
+import type { CategoryData, ItemCategoryData } from '../sync/operations/types.js';
+import { isBlobCategoryData, isItemCategoryData } from '../sync/operations/types.js';
 
 /**
  * Write synced data back to local filesystem.
+ * - Blob categories: overwrites local data
+ * - Item categories: merges new items (does NOT overwrite existing)
  */
 export async function writeLocalData(
   pathConfig: PathConfig,
   categories: CategoryData[]
 ): Promise<void> {
   const categoryPaths = getCategoryPaths(pathConfig);
 
-  for (const { category, data } of categories) {
-    const paths = categoryPaths[category];
+  for (const catData of categories) {
+    const paths = categoryPaths[catData.category];
     if (paths.length === 0) continue;
 
-    const parsed = JSON.parse(data) as Record<string, unknown>;
-    await writeCategoryData(paths, parsed);
+    if (isItemCategoryData(catData)) {
+      // Per-item merge write
+      await writeItemCategoryData(catData.category, paths, catData);
+    } else if (isBlobCategoryData(catData)) {
+      // Blob overwrite
+      const parsed = JSON.parse(catData.data) as Record<string, unknown>;
+      await writeBlobCategoryData(paths, parsed);
+    }
   }
 }
 
 /**
- * Write category data to filesystem.
+ * Write blob-based category data to filesystem (overwrites).
  */
-async function writeCategoryData(paths: string[], data: Record<string, unknown>): Promise<void> {
+async function writeBlobCategoryData(
+  paths: string[],
+  data: Record<string, unknown>
+): Promise<void> {
   for (const [key, value] of Object.entries(data)) {
     const targetPath = findTargetPath(paths, key);
     if (targetPath === undefined) continue;
     await writeEntry(targetPath, value);
   }
 }
 
+/**
+ * Write per-item category data to filesystem (merges).
+ * Only writes new items that were pulled from remote.
+ */
+async function writeItemCategoryData(
+  _category: SyncCategory,
+  paths: string[],
+  catData: ItemCategoryData
+): Promise<void> {
+  // Find the base directory for this category
+  const basePath = paths[0];
+  if (!basePath) return;
+
+  // Each item ID is like "session/abc123.json" or "message/xyz/msg.json"
+  // We need to write to the appropriate location
+  for (const [itemId, content] of Object.entries(catData.items)) {
+    const targetPath = resolveItemPath(basePath, itemId);
+    await writeItemFile(targetPath, content);
+  }
+}
+
+/**
+ * Resolve the full filesystem path for an item.
+ * Item IDs are like "session/abc123.json" → basePath + abc123.json
+ * Or "message/session_id/msg.json" → basePath + session_id/msg.json
+ */
+function resolveItemPath(basePath: string, itemId: string): string {
+  // Item ID format: "basedir/relative/path.json"
+  // We need to strip the first segment (which matches basePath's basename)
+  const parts = itemId.split('/');
+  parts.shift(); // Remove the first segment (e.g., "session" or "message")
+  const relativePath = parts.join('/');
+  return join(basePath, relativePath);
+}
+
+/**
+ * Write a single item file.
+ */
+async function writeItemFile(filePath: string, content: string): Promise<void> {
+  try {
+    await ensureParentDir(filePath);
+    await writeFile(filePath, content, 'utf-8');
+  } catch (error) {
+    console.error(`Failed to write item ${filePath}:`, error);
+  }
+}
+
 /**
  * Find target path for a key.
  */
@@ -104,10 +164,44 @@ function serializeContent(filePath: string, value: unknown): string {
  * Ensure parent directory exists.
  */
 async function ensureParentDir(filePath: string): Promise<void> {
-  const parts = filePath.split('/');
-  parts.pop();
-  const parentDir = parts.join('/');
+  const parentDir = dirname(filePath);
   if (parentDir) {
     await mkdir(parentDir, { recursive: true });
   }
 }
+
+/**
+ * Delete tombstoned items from local filesystem.
+ * Called after pull when remote has tombstones for items that should be deleted locally.
+ */
+export async function deleteTombstonedItems(
+  pathConfig: PathConfig,
+  tombstonedItems: Partial<Record<SyncCategory, string[]>>
+): Promise<void> {
+  const categoryPaths = getCategoryPaths(pathConfig);
+
+  for (const [category, itemIds] of Object.entries(tombstonedItems)) {
+    const paths = categoryPaths[category as SyncCategory];
+    const basePath = paths[0];
+    if (!basePath) continue;
+
+    for (const itemId of itemIds) {
+      const targetPath = resolveItemPath(basePath, itemId);
+      await deleteItemFile(targetPath);
+    }
+  }
+}
+
+/**
+ * Delete a single item file (silently ignores if not found).
+ */
+async function deleteItemFile(filePath: string): Promise<void> {
+  try {
+    await unlink(filePath);
+  } catch (error) {
+    // ENOENT = file doesn't exist, which is fine
+    if ((error as NodeJS.ErrnoException).code !== 'ENOENT') {
+      console.error(`Failed to delete item ${filePath}:`, error);
+    }
+  }
+}