Files
pikasTech-HWLAB/internal/project-management/store.ts
T

517 lines
20 KiB
TypeScript

import pg from "pg";
import { buildPostgresPoolConfig } from "../db/runtime-store.ts";
const { Pool } = pg;
const schemaStatements = Object.freeze([
`CREATE TABLE IF NOT EXISTS project_management_schema_migrations (
migration_id TEXT PRIMARY KEY,
applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`,
`CREATE TABLE IF NOT EXISTS project_sources (
source_id TEXT PRIMARY KEY,
kind TEXT NOT NULL,
display_name TEXT NOT NULL,
project_id TEXT NOT NULL,
root_ref TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'active',
source_json JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`,
`CREATE TABLE IF NOT EXISTS mdtodo_documents (
source_id TEXT NOT NULL REFERENCES project_sources(source_id) ON DELETE CASCADE,
file_ref TEXT NOT NULL,
project_id TEXT NOT NULL,
relative_path TEXT NOT NULL,
title TEXT NOT NULL,
fingerprint TEXT NOT NULL,
task_count INTEGER NOT NULL DEFAULT 0,
parse_error TEXT,
revision INTEGER NOT NULL DEFAULT 1,
document_json JSONB NOT NULL DEFAULT '{}'::jsonb,
last_indexed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (source_id, file_ref)
)`,
`CREATE TABLE IF NOT EXISTS mdtodo_task_projection (
task_ref TEXT PRIMARY KEY,
project_id TEXT NOT NULL,
source_id TEXT NOT NULL,
file_ref TEXT NOT NULL,
task_id TEXT NOT NULL,
title TEXT NOT NULL,
status TEXT NOT NULL,
parent_task_ref TEXT,
depth INTEGER NOT NULL DEFAULT 0,
line_number INTEGER NOT NULL DEFAULT 0,
ordinal INTEGER NOT NULL DEFAULT 0,
link_count INTEGER NOT NULL DEFAULT 0,
source_fingerprint TEXT NOT NULL,
task_json JSONB NOT NULL DEFAULT '{}'::jsonb,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`,
`CREATE TABLE IF NOT EXISTS project_workbench_links (
link_id TEXT PRIMARY KEY,
project_id TEXT NOT NULL,
task_ref TEXT,
session_id TEXT,
trace_id TEXT,
created_by TEXT,
role TEXT NOT NULL DEFAULT 'reference',
link_json JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`,
`CREATE TABLE IF NOT EXISTS project_management_audit_events (
event_id TEXT PRIMARY KEY,
actor_id TEXT,
action TEXT NOT NULL,
target_type TEXT NOT NULL,
target_id TEXT NOT NULL,
outcome TEXT NOT NULL,
event_json JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`,
`CREATE TABLE IF NOT EXISTS project_management_outbox (
event_id TEXT PRIMARY KEY,
aggregate_type TEXT NOT NULL,
aggregate_id TEXT NOT NULL,
event_type TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
event_json JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
published_at TIMESTAMPTZ
)`,
`CREATE INDEX IF NOT EXISTS idx_mdtodo_documents_project ON mdtodo_documents(project_id, relative_path)`,
`CREATE INDEX IF NOT EXISTS idx_mdtodo_tasks_project_file ON mdtodo_task_projection(project_id, source_id, file_ref, ordinal)`,
`CREATE INDEX IF NOT EXISTS idx_mdtodo_tasks_parent ON mdtodo_task_projection(parent_task_ref)`,
`CREATE INDEX IF NOT EXISTS idx_project_workbench_links_task ON project_workbench_links(task_ref, updated_at DESC)`,
`CREATE INDEX IF NOT EXISTS idx_project_outbox_status ON project_management_outbox(status, created_at)`
]);
export function createProjectManagementStore(options = {}) {
const env = options.env ?? process.env;
const kind = String(options.kind ?? env.HWLAB_PROJECT_MANAGEMENT_STORE ?? "postgres").trim().toLowerCase();
if (kind === "memory") return new MemoryProjectManagementStore(options);
const dbUrl = options.dbUrl ?? env.HWLAB_PROJECT_MANAGEMENT_DB_URL;
if (!dbUrl) return new UnconfiguredProjectManagementStore("HWLAB_PROJECT_MANAGEMENT_DB_URL");
return new PostgresProjectManagementStore({
...options,
dbUrl,
sslMode: options.sslMode ?? env.HWLAB_PROJECT_MANAGEMENT_DB_SSL_MODE,
timeoutMs: options.timeoutMs ?? env.HWLAB_PROJECT_MANAGEMENT_DB_CONNECT_TIMEOUT_MS,
poolMax: options.poolMax ?? env.HWLAB_PROJECT_MANAGEMENT_DB_POOL_MAX,
queryTimeoutMs: options.queryTimeoutMs ?? env.HWLAB_PROJECT_MANAGEMENT_DB_QUERY_TIMEOUT_MS
});
}
export class PostgresProjectManagementStore {
constructor(options = {}) {
this.pool = options.pool ?? new Pool(buildPostgresPoolConfig({
dbUrl: options.dbUrl,
sslMode: options.sslMode,
timeoutMs: options.timeoutMs,
poolMax: options.poolMax,
queryTimeoutMs: options.queryTimeoutMs
}));
this.schemaReady = false;
}
async ensureSchema() {
if (this.schemaReady) return;
const client = await this.pool.connect();
try {
await client.query("BEGIN");
for (const statement of schemaStatements) await client.query(statement);
await client.query(
"INSERT INTO project_management_schema_migrations (migration_id) VALUES ($1) ON CONFLICT (migration_id) DO NOTHING",
["project-management-20260625-p1"]
);
await client.query("COMMIT");
this.schemaReady = true;
} catch (error) {
await client.query("ROLLBACK").catch(() => {});
throw error;
} finally {
client.release();
}
}
async upsertSource(source) {
await this.ensureSchema();
await this.pool.query(
`INSERT INTO project_sources (source_id, kind, display_name, project_id, root_ref, status, source_json, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, now())
ON CONFLICT (source_id) DO UPDATE SET
kind = EXCLUDED.kind,
display_name = EXCLUDED.display_name,
project_id = EXCLUDED.project_id,
root_ref = EXCLUDED.root_ref,
status = EXCLUDED.status,
source_json = EXCLUDED.source_json,
updated_at = now()`,
[source.sourceId, source.kind, source.displayName, source.projectId, source.rootRef, source.status, JSON.stringify(source)]
);
}
async replaceProjection(source, documents) {
await this.ensureSchema();
const client = await this.pool.connect();
try {
await client.query("BEGIN");
await client.query("DELETE FROM mdtodo_task_projection WHERE source_id = $1", [source.sourceId]);
await client.query("DELETE FROM mdtodo_documents WHERE source_id = $1", [source.sourceId]);
for (const document of documents) {
await client.query(
`INSERT INTO mdtodo_documents (source_id, file_ref, project_id, relative_path, title, fingerprint, task_count, document_json, last_indexed_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, now(), now())`,
[source.sourceId, document.fileRef, document.projectId, document.relativePath, document.title, document.fingerprint, document.taskCount, JSON.stringify(document.document)]
);
for (const task of document.tasks) {
await client.query(
`INSERT INTO mdtodo_task_projection
(task_ref, project_id, source_id, file_ref, task_id, title, status, parent_task_ref, depth, line_number, ordinal, link_count, source_fingerprint, task_json, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14::jsonb, now())`,
[task.taskRef, task.projectId, task.sourceId, task.fileRef, task.taskId, task.title, task.status, task.parentTaskRef, task.depth, task.lineNumber, task.ordinal, task.linkCount, task.sourceFingerprint, JSON.stringify(task)]
);
}
}
await client.query("COMMIT");
} catch (error) {
await client.query("ROLLBACK").catch(() => {});
throw error;
} finally {
client.release();
}
}
async listSources() {
await this.ensureSchema();
const result = await this.pool.query(
`SELECT source_id, kind, display_name, project_id, root_ref, status, source_json, updated_at
FROM project_sources ORDER BY source_id`
);
return result.rows.map((row) => ({
sourceId: row.source_id,
kind: row.kind,
displayName: row.display_name,
projectId: row.project_id,
rootRef: row.root_ref,
status: row.status,
updatedAt: row.updated_at,
...(row.source_json ?? {})
}));
}
async getSource(sourceId) {
await this.ensureSchema();
const result = await this.pool.query(
`SELECT source_id, kind, display_name, project_id, root_ref, status, source_json, updated_at
FROM project_sources WHERE source_id = $1`,
[sourceId]
);
const row = result.rows[0];
if (!row) return null;
return {
sourceId: row.source_id,
kind: row.kind,
displayName: row.display_name,
projectId: row.project_id,
rootRef: row.root_ref,
status: row.status,
updatedAt: row.updated_at,
...(row.source_json ?? {})
};
}
async listDocuments(filters = {}) {
await this.ensureSchema();
const params = [];
const clauses = [];
if (filters.sourceId) {
params.push(filters.sourceId);
clauses.push(`source_id = $${params.length}`);
}
const result = await this.pool.query(
`SELECT source_id, file_ref, project_id, relative_path, title, fingerprint, task_count, parse_error, revision, document_json, last_indexed_at, updated_at
FROM mdtodo_documents ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""}
ORDER BY project_id, relative_path`,
params
);
return result.rows.map((row) => ({
sourceId: row.source_id,
fileRef: row.file_ref,
projectId: row.project_id,
relativePath: row.relative_path,
title: row.title,
fingerprint: row.fingerprint,
taskCount: row.task_count,
parseError: row.parse_error,
revision: row.revision,
lastIndexedAt: row.last_indexed_at,
updatedAt: row.updated_at,
document: row.document_json ?? {}
}));
}
async listTasks(filters = {}) {
await this.ensureSchema();
const params = [];
const clauses = [];
appendTaskFilterClauses(params, clauses, filters);
let sql = `SELECT task_ref, project_id, source_id, file_ref, task_id, title, status, parent_task_ref, depth, line_number, ordinal, link_count, source_fingerprint, task_json, updated_at
FROM mdtodo_task_projection ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""}
ORDER BY project_id, source_id, file_ref, ordinal`;
if (Number.isInteger(filters.limit) && filters.limit > 0) {
params.push(filters.limit);
sql += ` LIMIT $${params.length}`;
}
if (Number.isInteger(filters.offset) && filters.offset > 0) {
params.push(filters.offset);
sql += ` OFFSET $${params.length}`;
}
const result = await this.pool.query(
sql,
params
);
return result.rows.map(taskFromRow);
}
async countTasks(filters = {}) {
await this.ensureSchema();
const params = [];
const clauses = [];
appendTaskFilterClauses(params, clauses, filters);
const result = await this.pool.query(
`SELECT COUNT(*)::int AS total FROM mdtodo_task_projection ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""}`,
params
);
return Number(result.rows[0]?.total ?? 0);
}
async getTask(taskRef) {
return (await this.listTasks({ taskRef, limit: 1 }))[0] ?? null;
}
async listWorkbenchLinks(filters = {}) {
await this.ensureSchema();
const params = [];
const clauses = [];
for (const [column, value] of [["task_ref", filters.taskRef], ["project_id", filters.projectId], ["session_id", filters.sessionId]]) {
if (!value) continue;
params.push(value);
clauses.push(`${column} = $${params.length}`);
}
const result = await this.pool.query(
`SELECT link_id, project_id, task_ref, session_id, trace_id, created_by, role, link_json, created_at, updated_at
FROM project_workbench_links ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""}
ORDER BY updated_at DESC, link_id LIMIT 200`,
params
);
return result.rows.map(workbenchLinkFromRow);
}
async upsertWorkbenchLink(link) {
await this.ensureSchema();
const result = await this.pool.query(
`INSERT INTO project_workbench_links (link_id, project_id, task_ref, session_id, trace_id, created_by, role, link_json, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, now(), now())
ON CONFLICT (link_id) DO UPDATE SET
project_id = EXCLUDED.project_id,
task_ref = EXCLUDED.task_ref,
session_id = EXCLUDED.session_id,
trace_id = EXCLUDED.trace_id,
created_by = COALESCE(EXCLUDED.created_by, project_workbench_links.created_by),
role = EXCLUDED.role,
link_json = EXCLUDED.link_json,
updated_at = now()
RETURNING link_id, project_id, task_ref, session_id, trace_id, created_by, role, link_json, created_at, updated_at`,
[link.linkId, link.projectId, link.taskRef ?? null, link.sessionId ?? null, link.traceId ?? null, link.createdBy ?? null, link.role ?? "launch", JSON.stringify(link.link ?? {})]
);
if (link.taskRef) {
await this.pool.query(
`UPDATE mdtodo_task_projection
SET link_count = (SELECT COUNT(*)::int FROM project_workbench_links WHERE task_ref = $1), updated_at = now()
WHERE task_ref = $1`,
[link.taskRef]
);
}
return workbenchLinkFromRow(result.rows[0]);
}
async recordAuditEvent(event) {
await this.ensureSchema();
await this.pool.query(
`INSERT INTO project_management_audit_events (event_id, actor_id, action, target_type, target_id, outcome, event_json, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, now())`,
[event.eventId, event.actorId ?? null, event.action, event.targetType, event.targetId, event.outcome, JSON.stringify(event)]
);
}
async close() {
await this.pool.end();
}
}
class MemoryProjectManagementStore {
constructor() {
this.sources = new Map();
this.documents = [];
this.tasks = [];
this.links = [];
this.auditEvents = [];
}
async ensureSchema() {}
async close() {}
async upsertSource(source) {
this.sources.set(source.sourceId, source);
}
async replaceProjection(source, documents) {
this.documents = this.documents.filter((document) => document.sourceId !== source.sourceId);
this.tasks = this.tasks.filter((task) => task.sourceId !== source.sourceId);
this.documents.push(...documents.map((document) => ({ ...document, lastIndexedAt: new Date().toISOString(), updatedAt: new Date().toISOString() })));
this.tasks.push(...documents.flatMap((document) => document.tasks.map((task) => ({ ...task, updatedAt: new Date().toISOString(), task }))));
}
async listSources() { return [...this.sources.values()]; }
async getSource(sourceId) { return this.sources.get(sourceId) ?? null; }
async listDocuments(filters = {}) { return this.documents.filter((item) => !filters.sourceId || item.sourceId === filters.sourceId); }
async listTasks(filters = {}) {
const offset = Number.isInteger(filters.offset) && filters.offset > 0 ? filters.offset : 0;
const limit = Number.isInteger(filters.limit) && filters.limit > 0 ? filters.limit : null;
const rows = filterMemoryTasks(this.tasks, filters).sort(compareProjectTasks);
return limit ? rows.slice(offset, offset + limit) : rows.slice(offset);
}
async countTasks(filters = {}) {
return filterMemoryTasks(this.tasks, filters).length;
}
async getTask(taskRef) {
return this.tasks.find((task) => task.taskRef === taskRef) ?? null;
}
async listWorkbenchLinks(filters = {}) {
return this.links.filter((item) => (!filters.taskRef || item.taskRef === filters.taskRef) && (!filters.projectId || item.projectId === filters.projectId) && (!filters.sessionId || item.sessionId === filters.sessionId));
}
async upsertWorkbenchLink(link) {
const now = new Date().toISOString();
const existingIndex = this.links.findIndex((item) => item.linkId === link.linkId);
const existing = existingIndex >= 0 ? this.links[existingIndex] : null;
const saved = { ...existing, ...link, role: link.role ?? "launch", link: link.link ?? existing?.link ?? {}, createdAt: existing?.createdAt ?? now, updatedAt: now };
if (existingIndex >= 0) this.links.splice(existingIndex, 1, saved);
else this.links.unshift(saved);
if (link.taskRef) {
const count = this.links.filter((item) => item.taskRef === link.taskRef).length;
this.tasks = this.tasks.map((task) => task.taskRef === link.taskRef ? { ...task, linkCount: count, updatedAt: now } : task);
}
return saved;
}
async recordAuditEvent(event) { this.auditEvents.push({ ...event, createdAt: new Date().toISOString() }); }
async listAuditEvents() { return this.auditEvents.slice(); }
}
class UnconfiguredProjectManagementStore {
constructor(envName) {
this.envName = envName;
}
async ensureSchema() {
const error = new Error(`${this.envName} is required for hwlab-project-management`);
error.code = "project_management_db_unconfigured";
error.statusCode = 503;
throw error;
}
async close() {}
async upsertSource() { await this.ensureSchema(); }
async getSource() { await this.ensureSchema(); }
async replaceProjection() { await this.ensureSchema(); }
async listSources() { await this.ensureSchema(); }
async listDocuments() { await this.ensureSchema(); }
async listTasks() { await this.ensureSchema(); }
async countTasks() { await this.ensureSchema(); }
async getTask() { await this.ensureSchema(); }
async listWorkbenchLinks() { await this.ensureSchema(); }
async upsertWorkbenchLink() { await this.ensureSchema(); }
async recordAuditEvent() { await this.ensureSchema(); }
}
function workbenchLinkFromRow(row) {
return {
linkId: row.link_id,
projectId: row.project_id,
taskRef: row.task_ref,
sessionId: row.session_id,
traceId: row.trace_id,
createdBy: row.created_by,
role: row.role,
createdAt: row.created_at,
updatedAt: row.updated_at,
link: row.link_json ?? {}
};
}
function appendTaskFilterClauses(params, clauses, filters = {}) {
for (const [column, value] of [["task_ref", filters.taskRef], ["source_id", filters.sourceId], ["file_ref", filters.fileRef], ["project_id", filters.projectId], ["parent_task_ref", filters.parentTaskRef], ["status", filters.status]]) {
if (!value) continue;
params.push(value);
clauses.push(`${column} = $${params.length}`);
}
const search = String(filters.search ?? "").trim();
if (search) {
params.push(`%${escapeSqlLike(search)}%`);
clauses.push(`(task_id ILIKE $${params.length} ESCAPE '\\' OR title ILIKE $${params.length} ESCAPE '\\' OR task_ref ILIKE $${params.length} ESCAPE '\\')`);
}
}
function taskFromRow(row) {
const taskJson = row.task_json && typeof row.task_json === "object" && !Array.isArray(row.task_json) ? row.task_json : {};
return {
...taskJson,
taskRef: row.task_ref,
projectId: row.project_id,
sourceId: row.source_id,
fileRef: row.file_ref,
taskId: row.task_id,
title: row.title,
status: row.status,
parentTaskRef: row.parent_task_ref,
depth: row.depth,
lineNumber: row.line_number,
ordinal: row.ordinal,
linkCount: row.link_count,
sourceFingerprint: row.source_fingerprint,
updatedAt: row.updated_at,
task: taskJson
};
}
function filterMemoryTasks(tasks, filters = {}) {
const search = String(filters.search ?? "").trim().toLowerCase();
return tasks.filter((item) => {
if (filters.taskRef && item.taskRef !== filters.taskRef) return false;
if (filters.sourceId && item.sourceId !== filters.sourceId) return false;
if (filters.fileRef && item.fileRef !== filters.fileRef) return false;
if (filters.projectId && item.projectId !== filters.projectId) return false;
if (filters.parentTaskRef && item.parentTaskRef !== filters.parentTaskRef) return false;
if (filters.status && item.status !== filters.status) return false;
if (search && ![item.taskId, item.title, item.taskRef].some((value) => String(value ?? "").toLowerCase().includes(search))) return false;
return true;
});
}
function compareProjectTasks(a, b) {
return String(a.projectId ?? "").localeCompare(String(b.projectId ?? ""))
|| String(a.sourceId ?? "").localeCompare(String(b.sourceId ?? ""))
|| String(a.fileRef ?? "").localeCompare(String(b.fileRef ?? ""))
|| Number(a.ordinal ?? 0) - Number(b.ordinal ?? 0);
}
function escapeSqlLike(value) {
return String(value).replace(/[\\%_]/gu, "\\$&");
}