feat(#71): implement graph data API
Implemented three new API endpoints for knowledge graph visualization: 1. GET /api/knowledge/graph - Full knowledge graph - Returns all entries and links with optional filtering - Supports filtering by tags, status, and node count limit - Includes orphan detection (entries with no links) 2. GET /api/knowledge/graph/stats - Graph statistics - Total entries and links counts - Orphan entries detection - Average links per entry - Top 10 most connected entries - Tag distribution across entries 3. GET /api/knowledge/graph/:slug - Entry-centered subgraph - Returns graph centered on specific entry - Supports depth parameter (1-5) for traversal distance - Includes all connected nodes up to specified depth New Files: - apps/api/src/knowledge/graph.controller.ts - apps/api/src/knowledge/graph.controller.spec.ts Modified Files: - apps/api/src/knowledge/dto/graph-query.dto.ts (added GraphFilterDto) - apps/api/src/knowledge/entities/graph.entity.ts (extended with new types) - apps/api/src/knowledge/services/graph.service.ts (added new methods) - apps/api/src/knowledge/services/graph.service.spec.ts (added tests) - apps/api/src/knowledge/knowledge.module.ts (registered controller) - apps/api/src/knowledge/dto/index.ts (exported new DTOs) - docs/scratchpads/71-graph-data-api.md (implementation notes) Test Coverage: 21 tests (all passing) - 14 service tests including orphan detection, filtering, statistics - 7 controller tests for all three endpoints Follows TDD principles with tests written before implementation. All code quality gates passed (lint, typecheck, tests). Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Jason Woltje
301
apps/orchestrator/src/queue/queue.service.ts
Normal file
301
apps/orchestrator/src/queue/queue.service.ts
Normal file
@@ -0,0 +1,301 @@
|
||||
import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
import { Queue, Worker, Job } from 'bullmq';
|
||||
import { ValkeyService } from '../valkey/valkey.service';
|
||||
import type { TaskContext } from '../valkey/types';
|
||||
import type {
|
||||
QueuedTask,
|
||||
QueueStats,
|
||||
AddTaskOptions,
|
||||
RetryConfig,
|
||||
TaskProcessingResult,
|
||||
} from './types';
|
||||
|
||||
/**
|
||||
* Queue service for managing task queue with priority and retry logic
|
||||
*/
|
||||
@Injectable()
|
||||
export class QueueService implements OnModuleInit, OnModuleDestroy {
|
||||
private queue!: Queue<QueuedTask>;
|
||||
private worker!: Worker<QueuedTask, TaskProcessingResult>;
|
||||
private readonly queueName: string;
|
||||
private readonly retryConfig: RetryConfig;
|
||||
|
||||
constructor(
|
||||
private readonly valkeyService: ValkeyService,
|
||||
private readonly configService: ConfigService
|
||||
) {
|
||||
this.queueName = this.configService.get<string>(
|
||||
'orchestrator.queue.name',
|
||||
'orchestrator-tasks'
|
||||
);
|
||||
|
||||
this.retryConfig = {
|
||||
maxRetries: this.configService.get<number>(
|
||||
'orchestrator.queue.maxRetries',
|
||||
3
|
||||
),
|
||||
baseDelay: this.configService.get<number>(
|
||||
'orchestrator.queue.baseDelay',
|
||||
1000
|
||||
),
|
||||
maxDelay: this.configService.get<number>(
|
||||
'orchestrator.queue.maxDelay',
|
||||
60000
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
async onModuleInit(): Promise<void> {
|
||||
// Initialize BullMQ with Valkey connection
|
||||
const connection = {
|
||||
host: this.configService.get<string>('orchestrator.valkey.host', 'localhost'),
|
||||
port: this.configService.get<number>('orchestrator.valkey.port', 6379),
|
||||
password: this.configService.get<string>('orchestrator.valkey.password'),
|
||||
};
|
||||
|
||||
// Create queue
|
||||
this.queue = new Queue<QueuedTask>(this.queueName, {
|
||||
connection,
|
||||
defaultJobOptions: {
|
||||
removeOnComplete: {
|
||||
age: 3600, // Keep completed jobs for 1 hour
|
||||
count: 100, // Keep last 100 completed jobs
|
||||
},
|
||||
removeOnFail: {
|
||||
age: 86400, // Keep failed jobs for 24 hours
|
||||
count: 1000, // Keep last 1000 failed jobs
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
// Create worker
|
||||
this.worker = new Worker<QueuedTask, TaskProcessingResult>(
|
||||
this.queueName,
|
||||
async (job: Job<QueuedTask>) => {
|
||||
return this.processTask(job);
|
||||
},
|
||||
{
|
||||
connection,
|
||||
concurrency: this.configService.get<number>(
|
||||
'orchestrator.queue.concurrency',
|
||||
5
|
||||
),
|
||||
}
|
||||
);
|
||||
|
||||
// Setup error handlers
|
||||
this.worker.on('failed', async (job, err) => {
|
||||
if (job) {
|
||||
await this.handleTaskFailure(job.data.taskId, err);
|
||||
}
|
||||
});
|
||||
|
||||
this.worker.on('completed', async (job) => {
|
||||
if (job) {
|
||||
await this.handleTaskCompletion(job.data.taskId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async onModuleDestroy(): Promise<void> {
|
||||
await this.worker.close();
|
||||
await this.queue.close();
|
||||
}
|
||||
|
||||
/**
|
||||
* Add task to queue
|
||||
*/
|
||||
async addTask(
|
||||
taskId: string,
|
||||
context: TaskContext,
|
||||
options?: AddTaskOptions
|
||||
): Promise<void> {
|
||||
// Validate options
|
||||
const priority = options?.priority ?? 5;
|
||||
const maxRetries = options?.maxRetries ?? this.retryConfig.maxRetries;
|
||||
const delay = options?.delay ?? 0;
|
||||
|
||||
if (priority < 1 || priority > 10) {
|
||||
throw new Error('Priority must be between 1 and 10');
|
||||
}
|
||||
|
||||
if (maxRetries < 0) {
|
||||
throw new Error('maxRetries must be non-negative');
|
||||
}
|
||||
|
||||
const queuedTask: QueuedTask = {
|
||||
taskId,
|
||||
priority,
|
||||
retries: 0,
|
||||
maxRetries,
|
||||
context,
|
||||
};
|
||||
|
||||
// Add to BullMQ queue
|
||||
await this.queue.add(taskId, queuedTask, {
|
||||
priority: 10 - priority + 1, // BullMQ: lower number = higher priority, so invert
|
||||
attempts: maxRetries + 1, // +1 for initial attempt
|
||||
backoff: {
|
||||
type: 'custom',
|
||||
},
|
||||
delay,
|
||||
});
|
||||
|
||||
// Update task state in Valkey
|
||||
await this.valkeyService.updateTaskStatus(taskId, 'pending');
|
||||
|
||||
// Publish event
|
||||
await this.valkeyService.publishEvent({
|
||||
type: 'task.queued',
|
||||
timestamp: new Date().toISOString(),
|
||||
taskId,
|
||||
data: { priority },
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Get queue statistics
|
||||
*/
|
||||
async getStats(): Promise<QueueStats> {
|
||||
const counts = await this.queue.getJobCounts(
|
||||
'waiting',
|
||||
'active',
|
||||
'completed',
|
||||
'failed',
|
||||
'delayed'
|
||||
);
|
||||
|
||||
return {
|
||||
pending: counts.waiting || 0,
|
||||
active: counts.active || 0,
|
||||
completed: counts.completed || 0,
|
||||
failed: counts.failed || 0,
|
||||
delayed: counts.delayed || 0,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculate exponential backoff delay
|
||||
*/
|
||||
calculateBackoffDelay(
|
||||
attemptNumber: number,
|
||||
baseDelay: number,
|
||||
maxDelay: number
|
||||
): number {
|
||||
const delay = baseDelay * Math.pow(2, attemptNumber);
|
||||
return Math.min(delay, maxDelay);
|
||||
}
|
||||
|
||||
/**
|
||||
* Pause queue processing
|
||||
*/
|
||||
async pause(): Promise<void> {
|
||||
await this.queue.pause();
|
||||
}
|
||||
|
||||
/**
|
||||
* Resume queue processing
|
||||
*/
|
||||
async resume(): Promise<void> {
|
||||
await this.queue.resume();
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove task from queue
|
||||
*/
|
||||
async removeTask(taskId: string): Promise<void> {
|
||||
const job = await this.queue.getJob(taskId);
|
||||
if (job) {
|
||||
await job.remove();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Process task (called by worker)
|
||||
*/
|
||||
private async processTask(
|
||||
job: Job<QueuedTask>
|
||||
): Promise<TaskProcessingResult> {
|
||||
const { taskId } = job.data;
|
||||
|
||||
try {
|
||||
// Update task state to executing
|
||||
await this.valkeyService.updateTaskStatus(taskId, 'executing');
|
||||
|
||||
// Publish event
|
||||
await this.valkeyService.publishEvent({
|
||||
type: 'task.processing',
|
||||
timestamp: new Date().toISOString(),
|
||||
taskId,
|
||||
data: { attempt: job.attemptsMade + 1 },
|
||||
});
|
||||
|
||||
// Task processing will be handled by agent spawner
|
||||
// For now, just mark as processing
|
||||
return {
|
||||
success: true,
|
||||
metadata: {
|
||||
attempt: job.attemptsMade + 1,
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
// Handle retry logic
|
||||
const shouldRetry = job.attemptsMade < job.data.maxRetries;
|
||||
|
||||
if (shouldRetry) {
|
||||
// Calculate backoff delay for next retry
|
||||
const delay = this.calculateBackoffDelay(
|
||||
job.attemptsMade + 1,
|
||||
this.retryConfig.baseDelay,
|
||||
this.retryConfig.maxDelay
|
||||
);
|
||||
|
||||
// BullMQ will automatically retry with the backoff
|
||||
await job.updateData({
|
||||
...job.data,
|
||||
retries: job.attemptsMade + 1,
|
||||
});
|
||||
|
||||
await this.valkeyService.publishEvent({
|
||||
type: 'task.retry',
|
||||
timestamp: new Date().toISOString(),
|
||||
taskId,
|
||||
data: {
|
||||
attempt: job.attemptsMade + 1,
|
||||
nextDelay: delay,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle task failure
|
||||
*/
|
||||
private async handleTaskFailure(taskId: string, error: Error): Promise<void> {
|
||||
await this.valkeyService.updateTaskStatus(taskId, 'failed', undefined, error.message);
|
||||
|
||||
await this.valkeyService.publishEvent({
|
||||
type: 'task.failed',
|
||||
timestamp: new Date().toISOString(),
|
||||
taskId,
|
||||
error: error.message,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle task completion
|
||||
*/
|
||||
private async handleTaskCompletion(taskId: string): Promise<void> {
|
||||
await this.valkeyService.updateTaskStatus(taskId, 'completed');
|
||||
|
||||
await this.valkeyService.publishEvent({
|
||||
type: 'task.completed',
|
||||
timestamp: new Date().toISOString(),
|
||||
taskId,
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user