T1/Phase 1: scaffold Vite SPA + Hono API + Bun worker
- apps/web: Vite + React 19 + TanStack Router/Query + shadcn/ui - apps/api: Hono + Bun on :3001 with /api/health, /api/auth/*, /mcp stubs - apps/worker: Bun worker stub, DB connection, graceful SIGTERM - apps/web-legacy/: old Next.js code moved aside (preserved for T2-T8 reference) - Dockerfiles: api (Bun), worker (Bun), spa (multi-stage Caddy) - Caddyfile: serves dist + reverse-proxies /api/* + /mcp to api - docker-compose.yml: 4-service target (api, spa, db, worker) - packages/db/src/client.ts: shared Drizzle client for api + worker - db/client.ts: root-level alias for convenience Parent: t_e1cbd87d -> t_24c9c3fd (T0)
This commit is contained in:
Hermes
@@ -0,0 +1,145 @@
|
||||
import { createAdminClient } from '../pocketbase';
|
||||
import type { Agent, AgentTask } from '@project-e/shared';
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asAgent(record: Record<string, unknown>): Agent {
|
||||
return record as unknown as Agent;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asAgentTask(record: Record<string, unknown>): AgentTask {
|
||||
return record as unknown as AgentTask;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse @mentions from text content
|
||||
* Matches @agentname pattern
|
||||
*/
|
||||
export function parseMentions(content: string): string[] {
|
||||
const regex = /@(\w+)/g;
|
||||
const mentions: string[] = [];
|
||||
let match;
|
||||
|
||||
while ((match = regex.exec(content)) !== null) {
|
||||
mentions.push(match[1].toLowerCase());
|
||||
}
|
||||
|
||||
return [...new Set(mentions)]; // Deduplicate
|
||||
}
|
||||
|
||||
/**
|
||||
* Process @mentions in content and route to agents
|
||||
*/
|
||||
export async function processMentions(
|
||||
content: string,
|
||||
entityType: string,
|
||||
entityId: string,
|
||||
userId: string
|
||||
): Promise<AgentTask[]> {
|
||||
const mentionNames = parseMentions(content);
|
||||
if (mentionNames.length === 0) return [];
|
||||
|
||||
const pb = createAdminClient();
|
||||
const createdTasks: AgentTask[] = [];
|
||||
|
||||
for (const mentionName of mentionNames) {
|
||||
// Find agent by name (case-insensitive)
|
||||
const agentResults = await pb.collection('agents').getFullList({
|
||||
filter: 'status = "active"',
|
||||
});
|
||||
const agents = agentResults.map((r) => asAgent(r as unknown as Record<string, unknown>));
|
||||
|
||||
const agent = agents.find((a) => a.name.toLowerCase() === mentionName);
|
||||
if (!agent) continue;
|
||||
|
||||
// Extract instruction text after the @mention
|
||||
const mentionRegex = new RegExp(`@${mentionName}\\s+(.+?)(?=@\\w+|$)`, 'is');
|
||||
const mentionMatch = content.match(mentionRegex);
|
||||
const instruction = mentionMatch ? mentionMatch[1].trim() : '';
|
||||
|
||||
// Create agent task
|
||||
const agentTaskRecord = await pb.collection('agent_tasks').create({
|
||||
agent_id: agent.id,
|
||||
task_type: 'mention',
|
||||
input: {
|
||||
entity_type: entityType,
|
||||
entity_id: entityId,
|
||||
instruction,
|
||||
user_id: userId,
|
||||
},
|
||||
status: 'pending',
|
||||
});
|
||||
const agentTask = asAgentTask(agentTaskRecord as unknown as Record<string, unknown>);
|
||||
|
||||
createdTasks.push(agentTask);
|
||||
|
||||
// Queue webhook delivery to agent if it has a webhook URL in config
|
||||
const agentConfig = (agent.config || {}) as Record<string, unknown>;
|
||||
const webhookUrl = agentConfig.webhook_url as string | undefined;
|
||||
|
||||
if (webhookUrl && agent.api_key) {
|
||||
await pb.collection('queue_jobs').create({
|
||||
queue: 'agents',
|
||||
type: 'agent_mention',
|
||||
payload: {
|
||||
agent_task_id: agentTask.id,
|
||||
agent_id: agent.id,
|
||||
agent_webhook_url: webhookUrl,
|
||||
agent_api_key: agent.api_key,
|
||||
entity_type: entityType,
|
||||
entity_id: entityId,
|
||||
instruction,
|
||||
user_id: userId,
|
||||
},
|
||||
status: 'pending',
|
||||
attempts: 0,
|
||||
max_attempts: 3,
|
||||
scheduled_at: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return createdTasks;
|
||||
}
|
||||
|
||||
/**
|
||||
* Deliver @mention to agent via webhook
|
||||
*/
|
||||
export async function deliverAgentMention(payload: {
|
||||
agent_task_id: string;
|
||||
agent_id: string;
|
||||
agent_webhook_url: string;
|
||||
agent_api_key: string;
|
||||
entity_type: string;
|
||||
entity_id: string;
|
||||
instruction: string;
|
||||
user_id: string;
|
||||
}): Promise<{ success: boolean; statusCode?: number; responseBody?: string }> {
|
||||
const { agent_webhook_url, agent_api_key, ...body } = payload;
|
||||
|
||||
try {
|
||||
const response = await fetch(agent_webhook_url, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': `Bearer ${agent_api_key}`,
|
||||
'X-Agent-Task-Id': payload.agent_task_id,
|
||||
},
|
||||
body: JSON.stringify(body),
|
||||
signal: AbortSignal.timeout(30000), // 30 second timeout for agent work
|
||||
});
|
||||
|
||||
const responseBody = await response.text();
|
||||
|
||||
return {
|
||||
success: response.ok,
|
||||
statusCode: response.status,
|
||||
responseBody,
|
||||
};
|
||||
} catch (error) {
|
||||
return {
|
||||
success: false,
|
||||
responseBody: String(error),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
import { createPocketBaseClient, createAdminClient } from '../pocketbase';
|
||||
import type { Habit, HabitLog, HabitScoreConfig } from '@project-e/shared';
|
||||
|
||||
function asHabit(record: Record<string, unknown>): Habit {
|
||||
return record as unknown as Habit;
|
||||
}
|
||||
|
||||
function asHabitLog(record: Record<string, unknown>): HabitLog {
|
||||
return record as unknown as HabitLog;
|
||||
}
|
||||
|
||||
const DEFAULT_SCORE_CONFIG: Required<HabitScoreConfig> = {
|
||||
streak_weight: 0.5,
|
||||
consistency_weight: 0.2,
|
||||
difficulty_weight: 0.3,
|
||||
};
|
||||
|
||||
const DIFFICULTY_MULTIPLIER: Record<string, number> = {
|
||||
easy: 0.7,
|
||||
medium: 1.0,
|
||||
hard: 1.3,
|
||||
};
|
||||
|
||||
export function calculateHabitScore(
|
||||
habit: Habit,
|
||||
logs: HabitLog[],
|
||||
scoreConfig?: HabitScoreConfig
|
||||
): number {
|
||||
const config = { ...DEFAULT_SCORE_CONFIG, ...scoreConfig };
|
||||
const streakScore = Math.min((habit.current_streak / 30) * 100, 100);
|
||||
const thirtyDaysAgo = new Date();
|
||||
thirtyDaysAgo.setDate(thirtyDaysAgo.getDate() - 30);
|
||||
const recentLogs = logs.filter(
|
||||
(l) => new Date(l.logged_at) >= thirtyDaysAgo && (l as any).completed && !(l as any).skipped
|
||||
);
|
||||
const consistencyScore = (recentLogs.length / 30) * 100;
|
||||
const difficultyMultiplier = DIFFICULTY_MULTIPLIER[habit.difficulty] || 1.0;
|
||||
const difficultyScore = difficultyMultiplier * 100;
|
||||
const score = Math.round(
|
||||
streakScore * config.streak_weight +
|
||||
consistencyScore * config.consistency_weight +
|
||||
difficultyScore * config.difficulty_weight
|
||||
);
|
||||
return Math.min(Math.max(score, 0), 100);
|
||||
}
|
||||
|
||||
export function updateStreaks(
|
||||
habit: Habit,
|
||||
completionDate: string
|
||||
): { current_streak: number; best_streak: number } {
|
||||
const today = new Date(completionDate);
|
||||
const lastCompleted = habit.updated ? new Date(habit.updated) : null;
|
||||
let newStreak = habit.current_streak;
|
||||
if (lastCompleted) {
|
||||
const daysDiff = Math.floor((today.getTime() - lastCompleted.getTime()) / (1000 * 60 * 60 * 24));
|
||||
if (daysDiff === 1) newStreak = habit.current_streak + 1;
|
||||
else if (daysDiff === 0) newStreak = habit.current_streak;
|
||||
else newStreak = 1;
|
||||
} else {
|
||||
newStreak = 1;
|
||||
}
|
||||
return { current_streak: newStreak, best_streak: Math.max(newStreak, habit.best_streak) };
|
||||
}
|
||||
|
||||
export async function logHabitCompletion(
|
||||
habitId: string,
|
||||
data: { logged_at?: string; mood?: number; value?: number; notes?: string },
|
||||
token?: string
|
||||
): Promise<{ log: HabitLog; habit: Habit }> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
const loggedAt = data.logged_at || new Date().toISOString();
|
||||
const log = asHabitLog(
|
||||
await pb.collection('habit_logs').create({
|
||||
habit_id: habitId,
|
||||
mood: data.mood || undefined,
|
||||
notes: data.notes || '',
|
||||
logged_at: loggedAt,
|
||||
}) as unknown as Record<string, unknown>
|
||||
);
|
||||
const habit = asHabit(await pb.collection('habits').getOne(habitId) as unknown as Record<string, unknown>);
|
||||
const streaks = updateStreaks(habit, loggedAt);
|
||||
const allLogsResult = await pb.collection('habit_logs').getFullList({ filter: `habit_id = "${habitId}"` });
|
||||
const allLogs = allLogsResult.map((r) => asHabitLog(r as unknown as Record<string, unknown>));
|
||||
const score = calculateHabitScore(habit, allLogs, (habit as any).score_config as HabitScoreConfig);
|
||||
const updatedHabit = asHabit(
|
||||
await pb.collection('habits').update(habitId, {
|
||||
current_streak: streaks.current_streak,
|
||||
best_streak: streaks.best_streak,
|
||||
total_completions: (habit.total_completions || 0) + 1,
|
||||
}) as unknown as Record<string, unknown>
|
||||
);
|
||||
return { log, habit: updatedHabit };
|
||||
}
|
||||
|
||||
export function isHabitDueToday(habit: Habit): boolean {
|
||||
const today = new Date();
|
||||
const dayOfWeek = today.getDay();
|
||||
if ((habit as any).custom_days && (habit as any).custom_days.length > 0) {
|
||||
return (habit as any).custom_days.includes(dayOfWeek);
|
||||
}
|
||||
switch (habit.frequency) {
|
||||
case 'daily': return true;
|
||||
case 'weekly': {
|
||||
const startDate = (habit as any).start_date ? new Date((habit as any).start_date) : new Date(habit.created);
|
||||
return dayOfWeek === startDate.getDay();
|
||||
}
|
||||
case 'custom': return true;
|
||||
default: return true;
|
||||
}
|
||||
}
|
||||
|
||||
export async function getHabitsDueToday(token?: string): Promise<Habit[]> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
const results = await pb.collection('habits').getFullList({ filter: "archived != true" });
|
||||
return results.map((r) => asHabit(r as unknown as Record<string, unknown>)).filter((habit) => isHabitDueToday(habit));
|
||||
}
|
||||
|
||||
export async function getHabitStreaks(token?: string): Promise<Array<{ habit: Habit; current_streak: number; best_streak: number }>> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
const results = await pb.collection('habits').getFullList({ filter: "archived != true" });
|
||||
const habits = results.map((r) => asHabit(r as unknown as Record<string, unknown>));
|
||||
return habits
|
||||
.map((h) => ({ habit: h, current_streak: h.current_streak || 0, best_streak: h.best_streak || 0 }))
|
||||
.sort((a, b) => b.current_streak - a.current_streak);
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
export * from './task-service';
|
||||
export * from './habit-service';
|
||||
export * from './project-service';
|
||||
export * from './note-service';
|
||||
export * from './report-service';
|
||||
export * from './webhook-service';
|
||||
export * from './agent-mention-service';
|
||||
@@ -0,0 +1,307 @@
|
||||
import { createPocketBaseClient, createAdminClient } from '../pocketbase';
|
||||
import type { Note, NoteLink, NoteTaskLink, Task } from '@project-e/shared';
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asNote(record: Record<string, unknown>): Note {
|
||||
return record as unknown as Note;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asNoteLink(record: Record<string, unknown>): NoteLink {
|
||||
return record as unknown as NoteLink;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asNoteTaskLink(record: Record<string, unknown>): NoteTaskLink {
|
||||
return record as unknown as NoteTaskLink;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asTask(record: Record<string, unknown>): Task {
|
||||
return record as unknown as Task;
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract wikilinks from note content
|
||||
* Matches [[Note Title]] syntax
|
||||
*/
|
||||
export function extractWikilinks(content: string): string[] {
|
||||
const regex = /\[\[([^\]]+)\]\]/g;
|
||||
const links: string[] = [];
|
||||
let match;
|
||||
|
||||
while ((match = regex.exec(content)) !== null) {
|
||||
links.push(match[1].trim());
|
||||
}
|
||||
|
||||
return [...new Set(links)]; // Deduplicate
|
||||
}
|
||||
|
||||
/**
|
||||
* Update note_links when a note is saved
|
||||
* Re-syncs all outbound wikilinks from the note content
|
||||
*/
|
||||
export async function syncNoteLinks(
|
||||
noteId: string,
|
||||
content: string,
|
||||
token?: string
|
||||
): Promise<void> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
// Extract wikilinks from content
|
||||
const linkTitles = extractWikilinks(content);
|
||||
|
||||
// Get existing links from this note
|
||||
const existingResults = await pb.collection('note_links').getFullList({
|
||||
filter: `source_note_id = "${noteId}"`,
|
||||
});
|
||||
const existingLinks = existingResults.map((r) =>
|
||||
asNoteLink(r as unknown as Record<string, unknown>)
|
||||
);
|
||||
|
||||
// Delete existing links
|
||||
for (const link of existingLinks) {
|
||||
await pb.collection('note_links').delete(link.id);
|
||||
}
|
||||
|
||||
// Create new links
|
||||
for (const title of linkTitles) {
|
||||
// Find target note by title
|
||||
const targetNotes = await pb.collection('notes').getFullList({
|
||||
filter: `title = "${title}"`,
|
||||
});
|
||||
|
||||
if (targetNotes.length > 0) {
|
||||
await pb.collection('note_links').create({
|
||||
source_note_id: noteId,
|
||||
target_note_id: targetNotes[0].id,
|
||||
label: title,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get backlinks for a note (notes that link TO this note)
|
||||
*/
|
||||
export async function getBacklinks(
|
||||
noteId: string,
|
||||
token?: string
|
||||
): Promise<Note[]> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
// Find all note_links where this note is the target
|
||||
const linkResults = await pb.collection('note_links').getFullList({
|
||||
filter: `target_note_id = "${noteId}"`,
|
||||
});
|
||||
const links = linkResults.map((r) =>
|
||||
asNoteLink(r as unknown as Record<string, unknown>)
|
||||
);
|
||||
|
||||
// Get the source notes
|
||||
const backlinks: Note[] = [];
|
||||
for (const link of links) {
|
||||
const note = asNote(
|
||||
await pb
|
||||
.collection('notes')
|
||||
.getOne(link.source_note_id) as unknown as Record<string, unknown>
|
||||
);
|
||||
backlinks.push(note);
|
||||
}
|
||||
|
||||
return backlinks;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sync checkboxes in note content with tasks
|
||||
* Maps checkboxes to tasks via note_task_links
|
||||
*/
|
||||
export async function syncNoteTasks(
|
||||
noteId: string,
|
||||
content: string,
|
||||
token?: string
|
||||
): Promise<void> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
// Extract checkbox patterns: - [ ] Task title or - [x] Task title
|
||||
const checkboxRegex = /^- \[([ x])\] (.+)$/gm;
|
||||
const checkboxes: Array<{
|
||||
done: boolean;
|
||||
title: string;
|
||||
position: number;
|
||||
}> = [];
|
||||
let match;
|
||||
let position = 0;
|
||||
|
||||
while ((match = checkboxRegex.exec(content)) !== null) {
|
||||
checkboxes.push({
|
||||
done: match[1] === 'x',
|
||||
title: match[2].trim(),
|
||||
position: position++,
|
||||
});
|
||||
}
|
||||
|
||||
// Get existing note_task_links
|
||||
const existingResults = await pb.collection('note_task_links').getFullList({
|
||||
filter: `note_id = "${noteId}"`,
|
||||
});
|
||||
const existingLinks = existingResults.map((r) =>
|
||||
asNoteTaskLink(r as unknown as Record<string, unknown>)
|
||||
);
|
||||
|
||||
// Match checkboxes to existing links by index position
|
||||
for (let i = 0; i < checkboxes.length; i++) {
|
||||
const checkbox = checkboxes[i];
|
||||
const existingLink = existingLinks.find(
|
||||
(l) => l.label === `checkbox_${i}` || (!l.label && existingLinks.indexOf(l) === i)
|
||||
);
|
||||
|
||||
if (existingLink) {
|
||||
// Update existing task
|
||||
const task = asTask(
|
||||
await pb
|
||||
.collection('tasks')
|
||||
.getOne(existingLink.task_id) as unknown as Record<string, unknown>
|
||||
);
|
||||
|
||||
// Update title if changed
|
||||
if (task.title !== checkbox.title) {
|
||||
await pb.collection('tasks').update(existingLink.task_id, {
|
||||
title: checkbox.title,
|
||||
});
|
||||
}
|
||||
|
||||
// Update done status if changed
|
||||
const taskDone = task.status === 'done';
|
||||
if (taskDone !== checkbox.done) {
|
||||
await pb.collection('tasks').update(existingLink.task_id, {
|
||||
status: checkbox.done ? 'done' : 'todo',
|
||||
...(checkbox.done && { completed_at: new Date().toISOString() }),
|
||||
});
|
||||
}
|
||||
} else {
|
||||
// Create new task for this checkbox
|
||||
const note = asNote(
|
||||
await pb
|
||||
.collection('notes')
|
||||
.getOne(noteId) as unknown as Record<string, unknown>
|
||||
);
|
||||
|
||||
const newTask = await pb.collection('tasks').create({
|
||||
title: checkbox.title,
|
||||
description: '',
|
||||
status: checkbox.done ? 'done' : 'todo',
|
||||
priority: 'medium',
|
||||
domain: note.domain,
|
||||
tags: [],
|
||||
...(checkbox.done && { completed_at: new Date().toISOString() }),
|
||||
});
|
||||
|
||||
// Create mapping
|
||||
await pb.collection('note_task_links').create({
|
||||
note_id: noteId,
|
||||
task_id: newTask.id,
|
||||
label: `checkbox_${i}`,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Remove links for checkboxes that no longer exist
|
||||
for (const link of existingLinks) {
|
||||
const matchIndex = parseInt(
|
||||
(link.label || '').replace('checkbox_', ''),
|
||||
10
|
||||
);
|
||||
if (isNaN(matchIndex) || matchIndex >= checkboxes.length) {
|
||||
await pb.collection('note_task_links').delete(link.id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse frontmatter from note content
|
||||
* Frontmatter is YAML between --- delimiters at the top
|
||||
*/
|
||||
export function parseFrontmatter(content: string): {
|
||||
frontmatter: Record<string, unknown>;
|
||||
contentWithoutFrontmatter: string;
|
||||
} {
|
||||
const frontmatterRegex = /^---\n([\s\S]*?)\n---\n([\s\S]*)$/;
|
||||
const match = content.match(frontmatterRegex);
|
||||
|
||||
if (!match) {
|
||||
return { frontmatter: {}, contentWithoutFrontmatter: content };
|
||||
}
|
||||
|
||||
// Simple YAML parsing (key: value pairs)
|
||||
const yamlText = match[1];
|
||||
const frontmatter: Record<string, unknown> = {};
|
||||
|
||||
for (const line of yamlText.split('\n')) {
|
||||
const colonIndex = line.indexOf(':');
|
||||
if (colonIndex > 0) {
|
||||
const key = line.slice(0, colonIndex).trim();
|
||||
const value = line.slice(colonIndex + 1).trim();
|
||||
|
||||
// Try to parse as JSON for arrays/objects
|
||||
try {
|
||||
frontmatter[key] = JSON.parse(value);
|
||||
} catch {
|
||||
frontmatter[key] = value;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
frontmatter,
|
||||
contentWithoutFrontmatter: match[2],
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Get graph data for note visualization
|
||||
*/
|
||||
export async function getNoteGraph(
|
||||
token?: string
|
||||
): Promise<{
|
||||
nodes: Array<{
|
||||
id: string;
|
||||
title: string;
|
||||
domain: string;
|
||||
connectionCount: number;
|
||||
}>;
|
||||
edges: Array<{ source: string; target: string }>;
|
||||
}> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
const noteResults = await pb.collection('notes').getFullList();
|
||||
const notes = noteResults.map((r) => asNote(r as unknown as Record<string, unknown>));
|
||||
|
||||
const linkResults = await pb.collection('note_links').getFullList();
|
||||
const links = linkResults.map((r) =>
|
||||
asNoteLink(r as unknown as Record<string, unknown>)
|
||||
);
|
||||
|
||||
// Count connections per note
|
||||
const connectionCounts: Record<string, number> = {};
|
||||
for (const link of links) {
|
||||
connectionCounts[link.source_note_id] =
|
||||
(connectionCounts[link.source_note_id] || 0) + 1;
|
||||
connectionCounts[link.target_note_id] =
|
||||
(connectionCounts[link.target_note_id] || 0) + 1;
|
||||
}
|
||||
|
||||
const nodes = notes.map((n) => ({
|
||||
id: n.id,
|
||||
title: n.title,
|
||||
domain: n.domain,
|
||||
connectionCount: connectionCounts[n.id] || 0,
|
||||
}));
|
||||
|
||||
const edges = links.map((l) => ({
|
||||
source: l.source_note_id,
|
||||
target: l.target_note_id,
|
||||
}));
|
||||
|
||||
return { nodes, edges };
|
||||
}
|
||||
@@ -0,0 +1,140 @@
|
||||
import { createPocketBaseClient, createAdminClient } from '../pocketbase';
|
||||
import type { Project, Task, Milestone } from '@project-e/shared';
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asProject(record: Record<string, unknown>): Project {
|
||||
return record as unknown as Project;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asTask(record: Record<string, unknown>): Task {
|
||||
return record as unknown as Task;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asMilestone(record: Record<string, unknown>): Milestone {
|
||||
return record as unknown as Milestone;
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute project progress from task completion percentage
|
||||
* Returns the percentage but does not persist it (Project schema has no progress field)
|
||||
*/
|
||||
export async function computeProjectProgress(
|
||||
projectId: string,
|
||||
token?: string
|
||||
): Promise<number> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
// Get all tasks for this project
|
||||
const results = await pb.collection('tasks').getFullList({
|
||||
filter: `project_id = "${projectId}"`,
|
||||
});
|
||||
const tasks = results.map((r) => asTask(r as unknown as Record<string, unknown>));
|
||||
|
||||
if (tasks.length === 0) return 0;
|
||||
|
||||
const done = tasks.filter((t) => t.status === 'done').length;
|
||||
return Math.round((done / tasks.length) * 100);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get project with computed progress and task counts
|
||||
*/
|
||||
export async function getProjectWithProgress(
|
||||
projectId: string,
|
||||
token?: string
|
||||
): Promise<Project & { taskCount: number; doneCount: number; progress: number }> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
const project = asProject(
|
||||
await pb.collection('projects').getOne(projectId) as unknown as Record<string, unknown>
|
||||
);
|
||||
|
||||
const results = await pb.collection('tasks').getFullList({
|
||||
filter: `project_id = "${projectId}"`,
|
||||
});
|
||||
const tasks = results.map((r) => asTask(r as unknown as Record<string, unknown>));
|
||||
|
||||
const doneCount = tasks.filter((t) => t.status === 'done').length;
|
||||
const progress = tasks.length > 0 ? Math.round((doneCount / tasks.length) * 100) : 0;
|
||||
|
||||
return {
|
||||
...project,
|
||||
taskCount: tasks.length,
|
||||
doneCount,
|
||||
progress,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Check milestone dependency enforcement
|
||||
*/
|
||||
export async function canStartMilestone(
|
||||
milestoneId: string,
|
||||
token?: string
|
||||
): Promise<{ allowed: boolean; blockedBy: string[] }> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
// Get the milestone with its embedded dependencies
|
||||
const milestone = asMilestone(
|
||||
await pb.collection('milestones').getOne(milestoneId) as unknown as Record<string, unknown>
|
||||
);
|
||||
|
||||
const blockedBy: string[] = [];
|
||||
|
||||
for (const dep of milestone.dependencies || []) {
|
||||
const depMilestone = asMilestone(
|
||||
await pb
|
||||
.collection('milestones')
|
||||
.getOne(dep.depends_on_id) as unknown as Record<string, unknown>
|
||||
);
|
||||
if (depMilestone.status !== 'complete') {
|
||||
blockedBy.push(dep.depends_on_id);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
allowed: blockedBy.length === 0,
|
||||
blockedBy,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Get milestone timeline for a project
|
||||
*/
|
||||
export async function getMilestoneTimeline(
|
||||
projectId: string,
|
||||
token?: string
|
||||
): Promise<Milestone[]> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
const results = await pb.collection('milestones').getFullList({
|
||||
filter: `project_id = "${projectId}"`,
|
||||
sort: 'sort_order',
|
||||
});
|
||||
|
||||
return results.map((r) => asMilestone(r as unknown as Record<string, unknown>));
|
||||
}
|
||||
|
||||
/**
|
||||
* Log milestone status change to milestone_history
|
||||
*/
|
||||
export async function logMilestoneStatusChange(
|
||||
milestoneId: string,
|
||||
field: string,
|
||||
oldValue: string | undefined,
|
||||
newValue: string,
|
||||
changedBy?: string,
|
||||
token?: string
|
||||
): Promise<void> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
await pb.collection('milestone_history').create({
|
||||
milestone_id: milestoneId,
|
||||
field,
|
||||
old_value: oldValue,
|
||||
new_value: newValue,
|
||||
changed_by: changedBy,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,269 @@
|
||||
import { createAdminClient } from '../pocketbase';
|
||||
import type {
|
||||
Task,
|
||||
Habit,
|
||||
HabitLog,
|
||||
Milestone,
|
||||
TimeEntry,
|
||||
} from '@project-e/shared';
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asTask(record: Record<string, unknown>): Task {
|
||||
return record as unknown as Task;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asHabit(record: Record<string, unknown>): Habit {
|
||||
return record as unknown as Habit;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asHabitLog(record: Record<string, unknown>): HabitLog {
|
||||
return record as unknown as HabitLog;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asMilestone(record: Record<string, unknown>): Milestone {
|
||||
return record as unknown as Milestone;
|
||||
}
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asTimeEntry(record: Record<string, unknown>): TimeEntry {
|
||||
return record as unknown as TimeEntry;
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate weekly summary report data
|
||||
*/
|
||||
export async function generateWeeklySummary(
|
||||
weekStart: string,
|
||||
weekEnd: string,
|
||||
token?: string
|
||||
): Promise<{
|
||||
tasksCompleted: number;
|
||||
habitsTracked: number;
|
||||
timeLogged: number;
|
||||
streaks: Array<{ name: string; streak: number }>;
|
||||
byDomain: Record<string, { tasks: number; habits: number }>;
|
||||
}> {
|
||||
const pb = token ? createAdminClient() : createAdminClient();
|
||||
|
||||
// Tasks completed this week
|
||||
const taskResults = await pb.collection('tasks').getFullList({
|
||||
filter: `status = "done" && completed_at >= "${weekStart}" && completed_at <= "${weekEnd}"`,
|
||||
});
|
||||
const tasks = taskResults.map((r) => asTask(r as unknown as Record<string, unknown>));
|
||||
|
||||
// Habits logged this week
|
||||
const habitLogResults = await pb.collection('habit_logs').getFullList({
|
||||
filter: `logged_at >= "${weekStart}" && logged_at <= "${weekEnd}"`,
|
||||
});
|
||||
const habitLogs = habitLogResults.map((r) =>
|
||||
asHabitLog(r as unknown as Record<string, unknown>)
|
||||
);
|
||||
|
||||
// Time entries this week
|
||||
const timeEntryResults = await pb.collection('time_entries').getFullList({
|
||||
filter: `started_at >= "${weekStart}" && started_at <= "${weekEnd}"`,
|
||||
});
|
||||
const timeEntries = timeEntryResults.map((r) =>
|
||||
asTimeEntry(r as unknown as Record<string, unknown>)
|
||||
);
|
||||
|
||||
const totalTime = timeEntries.reduce(
|
||||
(sum, e) => sum + e.duration_minutes,
|
||||
0
|
||||
);
|
||||
|
||||
// Streaks
|
||||
const habitResults = await pb.collection('habits').getFullList();
|
||||
const habits = habitResults.map((r) => asHabit(r as unknown as Record<string, unknown>));
|
||||
const streaks = habits
|
||||
.map((h) => ({ name: h.name, streak: h.current_streak }))
|
||||
.sort((a, b) => b.streak - a.streak);
|
||||
|
||||
// By domain
|
||||
const byDomain: Record<string, { tasks: number; habits: number }> = {};
|
||||
for (const task of tasks) {
|
||||
if (!byDomain[task.domain])
|
||||
byDomain[task.domain] = { tasks: 0, habits: 0 };
|
||||
byDomain[task.domain].tasks++;
|
||||
}
|
||||
|
||||
return {
|
||||
tasksCompleted: tasks.length,
|
||||
habitsTracked: habitLogs.length,
|
||||
timeLogged: totalTime,
|
||||
streaks,
|
||||
byDomain,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate project health report data
|
||||
*/
|
||||
export async function generateProjectHealth(
|
||||
projectId: string,
|
||||
token?: string
|
||||
): Promise<{
|
||||
taskCount: number;
|
||||
doneCount: number;
|
||||
overdueCount: number;
|
||||
milestoneStatus: Record<string, number>;
|
||||
completionRate: number;
|
||||
}> {
|
||||
const pb = token ? createAdminClient() : createAdminClient();
|
||||
|
||||
const taskResults = await pb.collection('tasks').getFullList({
|
||||
filter: `project_id = "${projectId}"`,
|
||||
});
|
||||
const tasks = taskResults.map((r) => asTask(r as unknown as Record<string, unknown>));
|
||||
|
||||
const today = new Date().toISOString().split('T')[0];
|
||||
const doneCount = tasks.filter((t) => t.status === 'done').length;
|
||||
const overdueCount = tasks.filter(
|
||||
(t) => t.due_date && t.due_date < today && t.status !== 'done'
|
||||
).length;
|
||||
|
||||
// Milestone status
|
||||
const milestoneResults = await pb.collection('milestones').getFullList({
|
||||
filter: `project_id = "${projectId}"`,
|
||||
});
|
||||
const milestones = milestoneResults.map((r) =>
|
||||
asMilestone(r as unknown as Record<string, unknown>)
|
||||
);
|
||||
|
||||
const milestoneStatus: Record<string, number> = {
|
||||
planned: 0,
|
||||
in_progress: 0,
|
||||
complete: 0,
|
||||
};
|
||||
for (const m of milestones) {
|
||||
milestoneStatus[m.status] = (milestoneStatus[m.status] || 0) + 1;
|
||||
}
|
||||
|
||||
return {
|
||||
taskCount: tasks.length,
|
||||
doneCount,
|
||||
overdueCount,
|
||||
milestoneStatus,
|
||||
completionRate:
|
||||
tasks.length > 0 ? Math.round((doneCount / tasks.length) * 100) : 0,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate habit analysis report data
|
||||
*/
|
||||
export async function generateHabitAnalysis(
|
||||
days: number = 30,
|
||||
token?: string
|
||||
): Promise<{
|
||||
habits: Array<{
|
||||
id: string;
|
||||
name: string;
|
||||
streak: number;
|
||||
score: number;
|
||||
consistencyRate: number;
|
||||
completions: number;
|
||||
}>;
|
||||
atRiskHabits: string[];
|
||||
}> {
|
||||
const pb = token ? createAdminClient() : createAdminClient();
|
||||
|
||||
const startDate = new Date();
|
||||
startDate.setDate(startDate.getDate() - days);
|
||||
const startStr = startDate.toISOString().split('T')[0];
|
||||
|
||||
const habitResults = await pb.collection('habits').getFullList();
|
||||
const habits = habitResults.map((r) => asHabit(r as unknown as Record<string, unknown>));
|
||||
|
||||
const habitAnalysis = [];
|
||||
const atRiskHabits: string[] = [];
|
||||
|
||||
for (const habit of habits) {
|
||||
const logResults = await pb.collection('habit_logs').getFullList({
|
||||
filter: `habit_id = "${habit.id}" && logged_at >= "${startStr}" && completed = true && skipped = false`,
|
||||
});
|
||||
const logs = logResults.map((r) =>
|
||||
asHabitLog(r as unknown as Record<string, unknown>)
|
||||
);
|
||||
|
||||
const consistencyRate = Math.round((logs.length / days) * 100);
|
||||
|
||||
habitAnalysis.push({
|
||||
id: habit.id,
|
||||
name: habit.name,
|
||||
streak: habit.current_streak,
|
||||
score: habit.score,
|
||||
consistencyRate,
|
||||
completions: logs.length,
|
||||
});
|
||||
|
||||
// At risk: consistency < 50% or streak broken
|
||||
if (consistencyRate < 50 || habit.current_streak === 0) {
|
||||
atRiskHabits.push(habit.name);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
habits: habitAnalysis.sort((a, b) => b.score - a.score),
|
||||
atRiskHabits,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate time audit report data
|
||||
*/
|
||||
export async function generateTimeAudit(
|
||||
startDate: string,
|
||||
endDate: string,
|
||||
token?: string
|
||||
): Promise<{
|
||||
totalMinutes: number;
|
||||
byDomain: Record<string, number>;
|
||||
byProject: Record<string, number>;
|
||||
byTag: Record<string, number>;
|
||||
}> {
|
||||
const pb = token ? createAdminClient() : createAdminClient();
|
||||
|
||||
const entryResults = await pb.collection('time_entries').getFullList({
|
||||
filter: `started_at >= "${startDate}" && started_at <= "${endDate}"`,
|
||||
});
|
||||
const entries = entryResults.map((r) =>
|
||||
asTimeEntry(r as unknown as Record<string, unknown>)
|
||||
);
|
||||
|
||||
const byDomain: Record<string, number> = {};
|
||||
const byProject: Record<string, number> = {};
|
||||
const byTag: Record<string, number> = {};
|
||||
let totalMinutes = 0;
|
||||
|
||||
for (const entry of entries) {
|
||||
totalMinutes += entry.duration_minutes;
|
||||
|
||||
// Get the entity to access domain/project/tags
|
||||
// TimeEntry uses entity_type and entity_id
|
||||
if (entry.entity_type === 'task') {
|
||||
const task = asTask(
|
||||
await pb
|
||||
.collection('tasks')
|
||||
.getOne(entry.entity_id) as unknown as Record<string, unknown>
|
||||
);
|
||||
|
||||
byDomain[task.domain] =
|
||||
(byDomain[task.domain] || 0) + entry.duration_minutes;
|
||||
|
||||
if (task.project_id) {
|
||||
byProject[task.project_id] =
|
||||
(byProject[task.project_id] || 0) + entry.duration_minutes;
|
||||
}
|
||||
|
||||
for (const tag of task.tags || []) {
|
||||
byTag[tag] = (byTag[tag] || 0) + entry.duration_minutes;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return { totalMinutes, byDomain, byProject, byTag };
|
||||
}
|
||||
@@ -0,0 +1,176 @@
|
||||
import { createPocketBaseClient, createAdminClient } from '../pocketbase';
|
||||
import type { Task, Subtask } from '@project-e/shared';
|
||||
|
||||
/** Cast a PocketBase RecordModel to a typed domain model */
|
||||
function asTask(record: Record<string, unknown>): Task {
|
||||
return record as unknown as Task;
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculate task progress from subtask completion ratio
|
||||
*/
|
||||
export function calculateTaskProgress(subtasks: Subtask[]): number {
|
||||
if (subtasks.length === 0) return 0;
|
||||
const done = subtasks.filter((s) => s.done).length;
|
||||
return Math.round((done / subtasks.length) * 100);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a task is blocked by dependencies
|
||||
* Returns array of blocking task IDs
|
||||
*/
|
||||
export async function getBlockingDependencies(
|
||||
taskId: string,
|
||||
dependencies: string[]
|
||||
): Promise<string[]> {
|
||||
if (dependencies.length === 0) return [];
|
||||
|
||||
const pb = createAdminClient();
|
||||
const blocking: string[] = [];
|
||||
|
||||
for (const depId of dependencies) {
|
||||
const depTask = asTask(
|
||||
await pb.collection('tasks').getOne(depId)
|
||||
);
|
||||
if (depTask.status !== 'done') {
|
||||
blocking.push(depId);
|
||||
}
|
||||
}
|
||||
|
||||
return blocking;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check dependency gating — can this task be started?
|
||||
*/
|
||||
export async function canStartTask(
|
||||
taskId: string,
|
||||
dependencies: string[]
|
||||
): Promise<{ allowed: boolean; blockedBy: string[] }> {
|
||||
const blockedBy = await getBlockingDependencies(taskId, dependencies);
|
||||
return {
|
||||
allowed: blockedBy.length === 0,
|
||||
blockedBy,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle task completion — trigger recurring task spawn if needed
|
||||
*/
|
||||
export async function completeTask(
|
||||
taskId: string,
|
||||
token?: string
|
||||
): Promise<{ task: Task; nextRecurringTaskId?: string }> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
// Mark task as done
|
||||
const task = asTask(
|
||||
await pb.collection('tasks').update(taskId, {
|
||||
status: 'done',
|
||||
completed_at: new Date().toISOString(),
|
||||
})
|
||||
);
|
||||
|
||||
let nextRecurringTaskId: string | undefined;
|
||||
|
||||
// If recurring, spawn next occurrence
|
||||
if (task.recurring_config?.rule) {
|
||||
nextRecurringTaskId = await spawnNextRecurringTask(task);
|
||||
}
|
||||
|
||||
return { task, nextRecurringTaskId };
|
||||
}
|
||||
|
||||
/**
|
||||
* Spawn next recurring task from RRULE
|
||||
*/
|
||||
async function spawnNextRecurringTask(task: Task): Promise<string> {
|
||||
// Import rrule dynamically to avoid bundling issues
|
||||
const { RRule } = await import('rrule');
|
||||
|
||||
const rule = RRule.fromString(task.recurring_config!.rule);
|
||||
const now = new Date();
|
||||
const nextDate = rule.after(now, true);
|
||||
|
||||
if (!nextDate) {
|
||||
throw new Error('No next occurrence found for recurring task');
|
||||
}
|
||||
|
||||
const pb = createAdminClient();
|
||||
|
||||
// Create next occurrence
|
||||
const nextTask = await pb.collection('tasks').create({
|
||||
title: task.title,
|
||||
description: task.description || '',
|
||||
status: 'todo',
|
||||
priority: task.priority,
|
||||
due_date: nextDate.toISOString(),
|
||||
project_id: task.project_id || '',
|
||||
milestone_id: task.milestone_id || '',
|
||||
tags: task.tags || [],
|
||||
domain: task.domain,
|
||||
estimate: task.estimate || null,
|
||||
recurring_config: task.recurring_config,
|
||||
dependencies: task.dependencies || [],
|
||||
custom_fields: task.custom_fields || {},
|
||||
});
|
||||
|
||||
return nextTask.id;
|
||||
}
|
||||
|
||||
/**
|
||||
* Promote a subtask to a full task
|
||||
*/
|
||||
export async function promoteSubtask(
|
||||
parentTaskId: string,
|
||||
subtaskId: string,
|
||||
token?: string
|
||||
): Promise<Task> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
|
||||
// Get parent task
|
||||
const parentTask = asTask(
|
||||
await pb.collection('tasks').getOne(parentTaskId)
|
||||
);
|
||||
|
||||
// Find the subtask
|
||||
const subtasks = (parentTask.subtasks || []) as Subtask[];
|
||||
const subtask = subtasks.find((s) => s.id === subtaskId);
|
||||
|
||||
if (!subtask) {
|
||||
throw new Error('Subtask not found');
|
||||
}
|
||||
|
||||
// Create new task from subtask
|
||||
const newTask = await pb.collection('tasks').create({
|
||||
title: subtask.title,
|
||||
description: '',
|
||||
status: subtask.done ? 'done' : 'todo',
|
||||
priority: parentTask.priority,
|
||||
project_id: parentTask.project_id || '',
|
||||
domain: parentTask.domain,
|
||||
tags: parentTask.tags || [],
|
||||
});
|
||||
|
||||
// Remove subtask from parent
|
||||
const updatedSubtasks = subtasks.filter((s) => s.id !== subtaskId);
|
||||
await pb.collection('tasks').update(parentTaskId, {
|
||||
subtasks: updatedSubtasks,
|
||||
});
|
||||
|
||||
return asTask(newTask as unknown as Record<string, unknown>);
|
||||
}
|
||||
|
||||
/**
|
||||
* Auto-compute parent task progress from subtask completion
|
||||
*/
|
||||
export async function updateTaskProgress(
|
||||
taskId: string,
|
||||
token?: string
|
||||
): Promise<number> {
|
||||
const pb = token ? createPocketBaseClient(token) : createAdminClient();
|
||||
const task = asTask(await pb.collection('tasks').getOne(taskId));
|
||||
|
||||
const subtasks = (task.subtasks || []) as Subtask[];
|
||||
return calculateTaskProgress(subtasks);
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
import { createAdminClient } from '../pocketbase';
|
||||
import { eventBus, EVENTS } from '../events/event-bus';
|
||||
import type { Webhook } from '@project-e/shared';
|
||||
|
||||
/**
|
||||
* Initialize webhook service — subscribe to all events
|
||||
* Call this once at app startup
|
||||
*/
|
||||
export function initializeWebhookService(): void {
|
||||
// Subscribe to all domain events
|
||||
eventBus.on(EVENTS.TASK_COMPLETED, (data) => {
|
||||
queueWebhookDelivery('task.completed', data);
|
||||
});
|
||||
|
||||
eventBus.on(EVENTS.HABIT_COMPLETED, (data) => {
|
||||
queueWebhookDelivery('habit.completed', data);
|
||||
});
|
||||
|
||||
eventBus.on(EVENTS.HABIT_STREAK_BROKEN, (data) => {
|
||||
queueWebhookDelivery('habit.streak_broken', data);
|
||||
});
|
||||
|
||||
eventBus.on(EVENTS.MILESTONE_REACHED, (data) => {
|
||||
queueWebhookDelivery('milestone.reached', data);
|
||||
});
|
||||
|
||||
eventBus.on(EVENTS.PROJECT_STATUS_CHANGED, (data) => {
|
||||
queueWebhookDelivery('project.status_changed', data);
|
||||
});
|
||||
|
||||
eventBus.on(EVENTS.REPORT_GENERATED, (data) => {
|
||||
queueWebhookDelivery('report.generated', data);
|
||||
});
|
||||
|
||||
eventBus.on(EVENTS.AGENT_TASK_COMPLETED, (data) => {
|
||||
queueWebhookDelivery('agent_task.completed', data);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Queue a webhook delivery for all matching webhooks
|
||||
*/
|
||||
async function queueWebhookDelivery(eventType: string, payload: unknown): Promise<void> {
|
||||
const pb = createAdminClient();
|
||||
|
||||
try {
|
||||
// Get all active webhooks that subscribe to this event type
|
||||
const webhooks = await pb.collection('webhooks').getFullList({
|
||||
filter: 'active = true',
|
||||
}) as Webhook[];
|
||||
|
||||
const matchingWebhooks = webhooks.filter((webhook) => {
|
||||
const events = webhook.events as string[];
|
||||
return events.includes(eventType) || events.includes('*');
|
||||
});
|
||||
|
||||
// Queue delivery for each matching webhook
|
||||
for (const webhook of matchingWebhooks) {
|
||||
await pb.collection('queue_jobs').create({
|
||||
queue: 'webhooks',
|
||||
type: 'webhook_delivery',
|
||||
payload: {
|
||||
webhook_id: webhook.id,
|
||||
webhook_url: webhook.url,
|
||||
webhook_secret: webhook.secret || '',
|
||||
event_type: eventType,
|
||||
event_payload: payload,
|
||||
},
|
||||
status: 'pending',
|
||||
attempts: 0,
|
||||
max_attempts: webhook.retry_count || 3,
|
||||
scheduled_at: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Failed to queue webhook delivery:', error);
|
||||
// Log to error_logs collection
|
||||
await pb.collection('error_logs').create({
|
||||
level: 'error',
|
||||
source: 'webhook-service',
|
||||
message: 'Failed to queue webhook delivery',
|
||||
metadata: { eventType, payload, error: String(error) },
|
||||
}).catch(() => {
|
||||
// Ignore logging errors
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Deliver a webhook (called by the worker)
|
||||
*/
|
||||
export async function deliverWebhook(job: {
|
||||
webhook_id: string;
|
||||
webhook_url: string;
|
||||
webhook_secret: string;
|
||||
event_type: string;
|
||||
event_payload: unknown;
|
||||
}): Promise<{ success: boolean; statusCode?: number; responseBody?: string }> {
|
||||
const { webhook_url, webhook_secret, event_type, event_payload } = job;
|
||||
|
||||
try {
|
||||
// Create HMAC signature if secret is provided
|
||||
const headers: Record<string, string> = {
|
||||
'Content-Type': 'application/json',
|
||||
'X-Event-Type': event_type,
|
||||
};
|
||||
|
||||
if (webhook_secret) {
|
||||
const crypto = await import('node:crypto');
|
||||
const payload = JSON.stringify(event_payload);
|
||||
const signature = crypto
|
||||
.createHmac('sha256', webhook_secret)
|
||||
.update(payload)
|
||||
.digest('hex');
|
||||
headers['X-Webhook-Signature'] = signature;
|
||||
}
|
||||
|
||||
const response = await fetch(webhook_url, {
|
||||
method: 'POST',
|
||||
headers,
|
||||
body: JSON.stringify(event_payload),
|
||||
signal: AbortSignal.timeout(10000), // 10 second timeout
|
||||
});
|
||||
|
||||
const responseBody = await response.text();
|
||||
|
||||
return {
|
||||
success: response.ok,
|
||||
statusCode: response.status,
|
||||
responseBody,
|
||||
};
|
||||
} catch (error) {
|
||||
return {
|
||||
success: false,
|
||||
responseBody: String(error),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Record webhook delivery result
|
||||
*/
|
||||
export async function recordWebhookDelivery(
|
||||
webhookId: string,
|
||||
eventType: string,
|
||||
payload: unknown,
|
||||
result: { success: boolean; statusCode?: number; responseBody?: string },
|
||||
attempts: number
|
||||
): Promise<void> {
|
||||
const pb = createAdminClient();
|
||||
|
||||
await pb.collection('webhook_deliveries').create({
|
||||
webhook_id: webhookId,
|
||||
event: eventType,
|
||||
payload: payload as Record<string, unknown>,
|
||||
status: result.success ? 'success' : 'failed',
|
||||
status_code: result.statusCode || 0,
|
||||
response_body: result.responseBody || '',
|
||||
attempts,
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user