obsidian-plugin/taskgenius_taskgenius-plugin
1
0
Fork 0
mirror of https://github.com/taskgenius/taskgenius-plugin.git synced 2026-07-22 06:40:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(dataflow): resolve data persistence and task parsing issues

Browse source 
- Add initial scan when no cached data exists on startup
- Implement file modification time (mtime) tracking to skip unchanged files
- Fix Storage to use stable version (1.0.0) preventing cache invalidation
- Replace direct parsing with ConfigurableTaskParser for consistency
- Pass tgProject to parser ensuring tasks have correct project attributes
- Auto-persist index after batch operations preventing data loss
- Add comprehensive logging for debugging dataflow operations

Performance improvements:
- Skip parsing unchanged files based on mtime comparison
- Cache validation using both content hash and mtime
- Batch processing optimization with smart file skipping

Fixes data loss after restart and ensures proper task attributes.
This commit is contained in:
Quorafind 2025-08-19 23:36:56 +08:00
parent 4835367a4a
commit b84389ee95
3 changed files with 160 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

154
src/dataflow/Orchestrator.ts
View file

@ -1,5 +1,5 @@
import { App, TFile, Vault, MetadataCache } from "obsidian";
import type { Task } from "../types/task";
import type { Task, TgProject } from "../types/task";
import type { ProjectConfigManagerOptions } from "../managers/project-config-manager";
import { QueryAPI } from "./api/QueryAPI";
@ -17,6 +17,7 @@ import { ProjectDataWorkerManager } from "./workers/ProjectDataWorkerManager";
import { parseMarkdown } from "./parsers/MarkdownEntry";
import { parseCanvas } from "./parsers/CanvasEntry";
import { parseFileMeta } from "./parsers/FileMetaEntry";
import { ConfigurableTaskParser } from "./core/ConfigurableTaskParser";
/**
* DataflowOrchestrator - Coordinates all dataflow components
@ -68,6 +69,32 @@ export class DataflowOrchestrator {
async initialize(): Promise<void> {
await this.queryAPI.initialize();
// Check if we need to perform initial scan
const taskCount = (await this.queryAPI.getAllTasks()).length;
if (taskCount === 0) {
console.log("[DataflowOrchestrator] No cached tasks found, performing initial scan...");
// Get all markdown and canvas files
const mdFiles = this.vault.getMarkdownFiles();
const canvasFiles = this.vault.getFiles().filter(f => f.extension === "canvas");
const allFiles = [...mdFiles, ...canvasFiles];
console.log(`[DataflowOrchestrator] Found ${allFiles.length} files to process`);
// Process in batches for performance
const BATCH_SIZE = 50;
for (let i = 0; i < allFiles.length; i += BATCH_SIZE) {
const batch = allFiles.slice(i, i + BATCH_SIZE);
await this.processBatch(batch);
}
// Persist the initial index
await this.repository.persist();
const finalTaskCount = (await this.queryAPI.getAllTasks()).length;
console.log(`[DataflowOrchestrator] Initial scan complete, indexed ${finalTaskCount} tasks`);
}
// Initialize ObsidianSource to start listening for events
this.obsidianSource.initialize();
@ -105,25 +132,35 @@ export class DataflowOrchestrator {
const filePath = file.path;
try {
// Step 1: Check cache and parse if needed
// Step 1: Get file modification time
const fileStat = await this.vault.adapter.stat(filePath);
const mtime = fileStat?.mtime;
// Step 2: Check cache and parse if needed
const rawCached = await this.storage.loadRaw(filePath);
const fileContent = await this.vault.cachedRead(file);
let rawTasks: Task[];
if (rawCached && this.storage.isRawValid(filePath, rawCached, fileContent)) {
// Use cached raw tasks
let projectData: Awaited<ReturnType<typeof this.projectResolver.get>>;
if (rawCached && this.storage.isRawValid(filePath, rawCached, fileContent, mtime)) {
// Use cached raw tasks - file hasn't changed
console.log(`[DataflowOrchestrator] Using cached tasks for ${filePath} (mtime match)`);
rawTasks = rawCached.data;
// Still need to get project data for augmentation
projectData = await this.projectResolver.get(filePath);
} else {
// Parse the file
rawTasks = await this.parseFile(file);
console.log(`[DataflowOrchestrator] Parsing ${filePath} (cache miss or mtime mismatch)`);
// Store raw tasks with file content for hash
await this.storage.storeRaw(filePath, rawTasks, fileContent);
// Get project data first for parsing
projectData = await this.projectResolver.get(filePath);
rawTasks = await this.parseFile(file, projectData.tgProject);
// Store raw tasks with file content and mtime
await this.storage.storeRaw(filePath, rawTasks, fileContent, mtime);
}
// Step 2: Get project data (can be parallelized)
const projectData = await this.projectResolver.get(filePath);
// Store project data
await this.storage.storeProject(filePath, {
tgProject: projectData.tgProject,
@ -158,18 +195,36 @@ export class DataflowOrchestrator {
}
/**
* Parse a file based on its type
* Parse a file based on its type using ConfigurableTaskParser
*/
private async parseFile(file: TFile): Promise<Task[]> {
private async parseFile(file: TFile, tgProject?: TgProject): Promise<Task[]> {
const extension = file.extension.toLowerCase();
// Parse based on file type
let tasks: Task[] = [];
if (extension === "md") {
// Parse markdown tasks
// Use ConfigurableTaskParser for markdown files
const content = await this.vault.cachedRead(file);
const markdownTasks = await parseMarkdown(file.path, content);
const fileCache = this.metadataCache.getFileCache(file);
const fileMetadata = fileCache?.frontmatter || {};
// Create parser with plugin settings
const parser = new ConfigurableTaskParser({
parseMetadata: true,
parseTags: true,
parseComments: true,
parseHeadings: true,
customDateFormats: this.plugin.settings.customDateFormats,
statusMapping: this.plugin.settings.statusMapping || {},
emojiMapping: this.plugin.settings.emojiMapping || {},
specialTagPrefixes: this.plugin.settings.specialTagPrefixes || {},
fileMetadataInheritance: this.plugin.settings.fileMetadataInheritance,
projectConfig: this.plugin.settings.projectConfig
});
// Parse tasks using ConfigurableTaskParser with tgProject
const markdownTasks = parser.parseLegacy(content, file.path, fileMetadata, undefined, tgProject);
tasks.push(...markdownTasks);
// Parse file-level tasks from frontmatter
@ -190,35 +245,74 @@ export class DataflowOrchestrator {
*/
async processBatch(files: TFile[]): Promise<void> {
const updates = new Map<string, Task[]>();
let skippedCount = 0;
for (const file of files) {
try {
const filePath = file.path;
// Parse file
const rawTasks = await this.parseFile(file);
// Get file modification time
const fileStat = await this.vault.adapter.stat(file.path);
const mtime = fileStat?.mtime;
// Get project data
const projectData = await this.projectResolver.get(filePath);
// Check if we can skip this file based on cached data
const rawCached = await this.storage.loadRaw(filePath);
const fileContent = await this.vault.cachedRead(file);
// Augment tasks
const fileMetadata = this.metadataCache.getFileCache(file);
const augmentContext: AugmentContext = {
filePath,
fileMeta: fileMetadata?.frontmatter || {},
projectName: projectData.tgProject?.name,
projectMeta: projectData.enhancedMetadata,
tasks: rawTasks
};
const augmentedTasks = await this.augmentor.merge(augmentContext);
updates.set(filePath, augmentedTasks);
if (rawCached && this.storage.isRawValid(filePath, rawCached, fileContent, mtime)) {
// Skip this file - it hasn't changed
skippedCount++;
// Use cached raw tasks for augmentation
const rawTasks = rawCached.data;
// Get project data
const projectData = await this.projectResolver.get(filePath);
// Augment tasks
const fileMetadata = this.metadataCache.getFileCache(file);
const augmentContext: AugmentContext = {
filePath,
fileMeta: fileMetadata?.frontmatter || {},
projectName: projectData.tgProject?.name,
projectMeta: projectData.enhancedMetadata,
tasks: rawTasks
};
const augmentedTasks = await this.augmentor.merge(augmentContext);
updates.set(filePath, augmentedTasks);
} else {
// Parse file as it has changed or is new
// Get project data first for parsing
const projectData = await this.projectResolver.get(filePath);
const rawTasks = await this.parseFile(file, projectData.tgProject);
// Store raw tasks with mtime
await this.storage.storeRaw(filePath, rawTasks, fileContent, mtime);
// Augment tasks
const fileMetadata = this.metadataCache.getFileCache(file);
const augmentContext: AugmentContext = {
filePath,
fileMeta: fileMetadata?.frontmatter || {},
projectName: projectData.tgProject?.name,
projectMeta: projectData.enhancedMetadata,
tasks: rawTasks
};
const augmentedTasks = await this.augmentor.merge(augmentContext);
updates.set(filePath, augmentedTasks);
}
} catch (error) {
console.error(`Error processing file ${file.path} in batch:`, error);
}
}
if (skippedCount > 0) {
console.log(`[DataflowOrchestrator] Skipped ${skippedCount} unchanged files`);
}
// Update repository in batch
if (updates.size > 0) {
await this.repository.updateBatch(updates);

15
src/dataflow/indexer/Repository.ts
View file

@ -19,25 +19,34 @@ export class Repository {
private metadataCache: MetadataCache
) {
this.indexer = new TaskIndexer(app, vault, metadataCache);
this.storage = new Storage(app.appId || "obsidian-task-genius");
// Use a stable version string to avoid cache invalidation
this.storage = new Storage(app.appId || "obsidian-task-genius", "1.0.0");
}
/**
* Initialize the repository (load persisted data if available)
*/
async initialize(): Promise<void> {
console.log("[Repository] Initializing repository...");
// Try to load consolidated index from storage
const consolidated = await this.storage.loadConsolidated();
if (consolidated && consolidated.data) {
// Restore the index from persisted data
console.log("[Repository] Restoring index from persisted snapshot...");
await this.indexer.restoreFromSnapshot(consolidated.data);
const taskCount = await this.indexer.getTotalTaskCount();
console.log(`[Repository] Index restored with ${taskCount} tasks`);
// Emit cache ready event
emit(this.app, Events.CACHE_READY, {
initial: true,
timestamp: Date.now(),
seq: Seq.next()
});
} else {
console.log("[Repository] No persisted data found, starting with empty index");
}
}
@ -79,6 +88,10 @@ export class Repository {
totalChanged += tasks.length;
}
// Persist the consolidated index after batch updates
await this.persist();
console.log(`[Repository] Persisted index after batch update of ${changedFiles.length} files`);
// Emit batch update event
this.lastSequence = Seq.next();
emit(this.app, Events.TASK_CACHE_UPDATED, {

28
src/dataflow/persistence/Storage.ts
View file

@ -10,6 +10,7 @@ export interface RawRecord {
version: string;
schema: number;
data: Task[];
mtime?: number; // File modification time
}
export interface ProjectRecord {
@ -62,8 +63,9 @@ export class Storage {
private schemaVersion: number = 1;
constructor(appId: string, version?: string) {
this.currentVersion = version || "unknown";
this.currentVersion = version || "1.0.0"; // Use stable version instead of "unknown"
this.cache = new LocalStorageCache(appId, this.currentVersion);
console.log(`[Storage] Initialized with appId: ${appId}, version: ${this.currentVersion}`);
}
/**
@ -113,24 +115,32 @@ export class Storage {
/**
* Store raw tasks for a file
*/
async storeRaw(path: string, tasks: Task[], fileContent?: string): Promise<void> {
async storeRaw(path: string, tasks: Task[], fileContent?: string, mtime?: number): Promise<void> {
const record: RawRecord = {
hash: this.generateHash(fileContent || tasks),
time: Date.now(),
version: this.currentVersion,
schema: this.schemaVersion,
data: tasks,
mtime: mtime, // Store file modification time
};
await this.cache.storeFile(Keys.raw(path), record);
}
/**
* Check if raw tasks are valid based on content hash
* Check if raw tasks are valid based on content hash and modification time
*/
isRawValid(path: string, record: RawRecord, fileContent?: string): boolean {
isRawValid(path: string, record: RawRecord, fileContent?: string, mtime?: number): boolean {
if (!this.isVersionValid(record)) return false;
// Check modification time if provided
if (mtime !== undefined && record.mtime !== undefined) {
if (record.mtime !== mtime) {
return false; // File has been modified
}
}
// If file content provided, check hash
if (fileContent) {
const expectedHash = this.generateHash(fileContent);
@ -218,17 +228,22 @@ export class Storage {
async loadConsolidated(): Promise<ConsolidatedRecord | null> {
try {
const cached = await this.cache.loadConsolidatedCache<ConsolidatedRecord>('taskIndex');
if (!cached || !cached.data) return null;
if (!cached || !cached.data) {
console.log("[Storage] No consolidated cache found");
return null;
}
// Check version compatibility
if (!this.isVersionValid(cached.data)) {
console.log("[Storage] Consolidated cache version mismatch, clearing...");
await this.cache.removeFile(Keys.consolidated());
return null;
}
console.log(`[Storage] Loaded consolidated cache with ${cached.data.data ? Object.keys(cached.data.data).length : 0} entries`);
return cached.data;
} catch (error) {
console.error("Error loading consolidated index:", error);
console.error("[Storage] Error loading consolidated index:", error);
return null;
}
}
@ -244,6 +259,7 @@ export class Storage {
data: taskCache,
};
console.log(`[Storage] Storing consolidated cache with ${taskCache ? Object.keys(taskCache).length : 0} entries`);
await this.cache.storeConsolidatedCache('taskIndex', record);
}