diff --git a/bun.lock b/bun.lock index 3717409..a235ffd 100644 --- a/bun.lock +++ b/bun.lock @@ -5,6 +5,7 @@ "": { "name": "agentboard-init", "dependencies": { + "@vercel/blob": "^2.4.0", "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "lucide-react": "^0.576.0", @@ -495,6 +496,8 @@ "@unrs/resolver-binding-win32-x64-msvc": ["@unrs/resolver-binding-win32-x64-msvc@1.11.1", "", { "os": "win32", "cpu": "x64" }, "sha512-lrW200hZdbfRtztbygyaq/6jP6AKE8qQN2KvPcJ+x7wiD038YtnYtZ82IMNJ69GJibV7bwL3y9FgK+5w/pYt6g=="], + "@vercel/blob": ["@vercel/blob@2.4.0", "", { "dependencies": { "async-retry": "^1.3.3", "is-buffer": "^2.0.5", "is-node-process": "^1.2.0", "throttleit": "^2.1.0", "undici": "^6.23.0" } }, "sha512-ncQ8CRb6XoEAYJwjOTRGpACRT6h/AeY+/33gLyeVxG5BIes27OPm1jmqreF+JHjcTmGhClTP+kBpmyLfbV0xew=="], + "accepts": ["accepts@2.0.0", "", { "dependencies": { "mime-types": "^3.0.0", "negotiator": "^1.0.0" } }, "sha512-5cvg6CtKwfgdmVqY1WIiXKc3Q1bkRqGLi+2W/6ao+6Y7gu/RCwRuAhGEzh5B4KlszSuTLgZYuqFqo5bImjNKng=="], "acorn": ["acorn@8.16.0", "", { "bin": { "acorn": "bin/acorn" } }, "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw=="], @@ -541,6 +544,8 @@ "async-function": ["async-function@1.0.0", "", {}, "sha512-hsU18Ae8CDTR6Kgu9DYf0EbCr/a5iGL0rytQDobUcdpYOKokk8LEjVphnXkDkgpi0wYVsqrXuP0bZxJaTqdgoA=="], + "async-retry": ["async-retry@1.3.3", "", { "dependencies": { "retry": "0.13.1" } }, "sha512-wfr/jstw9xNi/0teMHrRW7dsz3Lt5ARhYNZ2ewpadnhaIp5mbALhOAP+EAdsC7t4Z6wqsDVv9+W6gm1Dk9mEyw=="], + "available-typed-arrays": ["available-typed-arrays@1.0.7", "", { "dependencies": { "possible-typed-array-names": "^1.0.0" } }, "sha512-wvUjBtSGN7+7SjNpq/9M2Tg350UZD3q62IFZLbRAR1bSMlCo1ZaeW+BJ+D090e4hIIZLBcTDWe4Mh4jvUDajzQ=="], "axe-core": ["axe-core@4.11.1", "", {}, "sha512-BASOg+YwO2C+346x3LZOeoovTIoTrRqEsqMa6fmfAV0P+U9mFr9NsyOEpiYvFjbc64NMrSswhV50WdXzdb/Z5A=="], @@ -885,6 +890,8 @@ "is-boolean-object": ["is-boolean-object@1.2.2", "", { "dependencies": { "call-bound": "^1.0.3", "has-tostringtag": "^1.0.2" } }, "sha512-wa56o2/ElJMYqjCjGkXri7it5FbebW5usLw/nPmCMs5DeZ7eziSYZhSmPRn0txqeW4LnAmQQU7FgqLpsEFKM4A=="], + "is-buffer": ["is-buffer@2.0.5", "", {}, "sha512-i2R6zNFDwgEHJyQUtJEk0XFi1i0dPFn/oqjK3/vPCcDeJvW5NQ83V8QbicfF1SupOaB0h8ntgBC2YiE7dfyctQ=="], + "is-bun-module": ["is-bun-module@2.0.0", "", { "dependencies": { "semver": "^7.7.1" } }, "sha512-gNCGbnnnnFAUGKeZ9PdbyeGYJqewpmc2aKHUEMO5nQPWU9lOmv7jcmQIv+qHD8fXW6W7qfuCwX4rY9LNRjXrkQ=="], "is-callable": ["is-callable@1.2.7", "", {}, "sha512-1BC0BVFhS/p0qtw6enp8e+8OD0UrK0oFLztSjNzhcKA3WDuJxxAPXzPuPtKkjEY9UUoEWlX/8fgKeu2S8i9JTA=="], @@ -1213,6 +1220,8 @@ "restore-cursor": ["restore-cursor@5.1.0", "", { "dependencies": { "onetime": "^7.0.0", "signal-exit": "^4.1.0" } }, "sha512-oMA2dcrw6u0YfxJQXm342bFKX/E4sG9rbTzO9ptUcR/e8A33cHuvStiYOwH7fszkZlZ1z/ta9AAoPk2F4qIOHA=="], + "retry": ["retry@0.13.1", "", {}, "sha512-XQBQ3I8W1Cge0Seh+6gjj03LbmRFWuoszgK9ooCpwYIrhhoO80pfq4cUkU5DkknwfOfFteRwlZ56PYOGYyFWdg=="], + "rettime": ["rettime@0.10.1", "", {}, "sha512-uyDrIlUEH37cinabq0AX4QbgV4HbFZ/gqoiunWQ1UqBtRvTTytwhNYjE++pO/MjPTZL5KQCf2bEoJ/BJNVQ5Kw=="], "reusify": ["reusify@1.1.0", "", {}, "sha512-g6QUff04oZpHs0eG5p83rFLhHeV00ug/Yf9nZM6fLeUrPguBTkTQOdpAWWspMh55TZfVQDPaN3NQJfbVRAxdIw=="], @@ -1319,6 +1328,8 @@ "tapable": ["tapable@2.3.0", "", {}, "sha512-g9ljZiwki/LfxmQADO3dEY1CbpmXT5Hm2fJ+QaGKwSXUylMybePR7/67YW7jOrrvjEgL1Fmz5kzyAjWVWLlucg=="], + "throttleit": ["throttleit@2.1.0", "", {}, "sha512-nt6AMGKW1p/70DF/hGBdJB57B8Tspmbp5gfJ8ilhLnt7kkr2ye7hzD6NVG8GGErk2HWF34igrL2CXmNIkzKqKw=="], + "tiny-invariant": ["tiny-invariant@1.3.3", "", {}, "sha512-+FbBPE1o9QAYvviau/qC5SE3caw21q3xkvWKBtja5vgqOWIHHJ3ioaq1VPfn/Szqctz2bU/oYeKd9/z5BL+PVg=="], "tinyexec": ["tinyexec@1.0.2", "", {}, "sha512-W/KYk+NFhkmsYpuHq5JykngiOCnxeVL8v8dFnqxSD8qEEdRfXk1SDM6JzNqcERbcGYj9tMrDQBYV9cjgnunFIg=="], @@ -1365,6 +1376,8 @@ "unbox-primitive": ["unbox-primitive@1.1.0", "", { "dependencies": { "call-bound": "^1.0.3", "has-bigints": "^1.0.2", "has-symbols": "^1.1.0", "which-boxed-primitive": "^1.1.1" } }, "sha512-nWJ91DjeOkej/TA8pXQ3myruKpKEYgqvpw9lz4OPHj/NWFNluYrjbz9j01CJ8yKQd2g4jFoOkINCTW2I5LEEyw=="], + "undici": ["undici@6.26.0", "", {}, "sha512-4yqz8a3n5HmGTlsbADNtr/dJlhkh/55Rq798G6ibiULcXbDtaLpTl1pvdqcbFfeoj3iSi52lePFM7h9H21cw/A=="], + "undici-types": ["undici-types@6.21.0", "", {}, "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ=="], "unicorn-magic": ["unicorn-magic@0.3.0", "", {}, "sha512-+QBBXBCvifc56fsbuxZQ6Sic3wqqc3WWaqxs58gvJrcOuN83HGTCwz3oS5phzU9LthRNE9VrJCFCLUgHeeFnfA=="], diff --git a/package.json b/package.json index 5827514..cf9bcf1 100644 --- a/package.json +++ b/package.json @@ -20,6 +20,7 @@ "test": "bun test" }, "dependencies": { + "@vercel/blob": "^2.4.0", "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "lucide-react": "^0.576.0", diff --git a/src/app/api/boards/[boardId]/activity/route.ts b/src/app/api/boards/[boardId]/activity/route.ts index 4e134ac..3c3fc4a 100644 --- a/src/app/api/boards/[boardId]/activity/route.ts +++ b/src/app/api/boards/[boardId]/activity/route.ts @@ -12,7 +12,7 @@ export async function GET(request: NextRequest, { params }: Params) { const limitRaw = url.searchParams.get("limit"); const limit = limitRaw ? Number.parseInt(limitRaw, 10) : undefined; - const storage = getStorage(); + const storage = await getStorage(); const events = await storage.listActivity(boardId, { limit: Number.isFinite(limit) ? limit : undefined, initiativeId: url.searchParams.get("initiativeId") || undefined, diff --git a/src/app/api/boards/[boardId]/agents/[agentId]/heartbeat/route.ts b/src/app/api/boards/[boardId]/agents/[agentId]/heartbeat/route.ts index f7bbad2..5aba331 100644 --- a/src/app/api/boards/[boardId]/agents/[agentId]/heartbeat/route.ts +++ b/src/app/api/boards/[boardId]/agents/[agentId]/heartbeat/route.ts @@ -14,7 +14,7 @@ export async function POST(request: NextRequest, { params }: Params) { } catch { // empty body is fine } - const storage = getStorage(); + const storage = await getStorage(); const agent = await storage.heartbeatAgent(boardId, agentId, message); if (!agent) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Agent not found" } }, { status: 404 }); diff --git a/src/app/api/boards/[boardId]/agents/[agentId]/route.ts b/src/app/api/boards/[boardId]/agents/[agentId]/route.ts index 7a066a7..bc0e9e5 100644 --- a/src/app/api/boards/[boardId]/agents/[agentId]/route.ts +++ b/src/app/api/boards/[boardId]/agents/[agentId]/route.ts @@ -8,7 +8,7 @@ type Params = { params: Promise<{ boardId: string; agentId: string }> }; export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId, agentId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const agent = await storage.getAgent(boardId, agentId); if (!agent) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Agent not found" } }, { status: 404 }); @@ -36,7 +36,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { if ("currentInitiativeId" in body) updates.currentInitiativeId = body.currentInitiativeId; if ("metadata" in body) updates.metadata = body.metadata; - const storage = getStorage(); + const storage = await getStorage(); const agent = await storage.updateAgent(boardId, agentId, updates); if (!agent) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Agent not found" } }, { status: 404 }); @@ -52,7 +52,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { export async function DELETE(_request: NextRequest, { params }: Params) { try { const { boardId, agentId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const agent = await storage.getAgent(boardId, agentId); const deleted = await storage.deleteAgent(boardId, agentId); if (!deleted) { diff --git a/src/app/api/boards/[boardId]/agents/route.ts b/src/app/api/boards/[boardId]/agents/route.ts index cfc59ce..7245ef4 100644 --- a/src/app/api/boards/[boardId]/agents/route.ts +++ b/src/app/api/boards/[boardId]/agents/route.ts @@ -242,7 +242,7 @@ export async function POST(request: NextRequest, { params }: Params) { ); } - const storage = getStorage(); + const storage = await getStorage(); const board = await storage.getBoard(boardId); if (!board) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Board not found" } }, { status: 404 }); @@ -310,7 +310,7 @@ export async function POST(request: NextRequest, { params }: Params) { export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const agents = await storage.listAgents(boardId); return NextResponse.json({ ok: true, data: agents }); } catch (err) { diff --git a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/route.ts b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/route.ts index 221a8f3..7b149d0 100644 --- a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/route.ts +++ b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/route.ts @@ -8,7 +8,7 @@ type Params = { params: Promise<{ boardId: string; initiativeId: string; planId: export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId, initiativeId, planId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const plan = await storage.getPlan(boardId, initiativeId, planId); if (!plan) { @@ -39,7 +39,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { if ("ownerAgentId" in body) updates.ownerAgentId = body.ownerAgentId; if ("tags" in body) updates.tags = body.tags; - const storage = getStorage(); + const storage = await getStorage(); const plan = await storage.updatePlan(boardId, initiativeId, planId, updates); if (!plan) { @@ -60,7 +60,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { export async function DELETE(_request: NextRequest, { params }: Params) { try { const { boardId, initiativeId, planId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const plan = await storage.getPlan(boardId, initiativeId, planId); const deleted = await storage.deletePlan(boardId, initiativeId, planId); diff --git a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/steps/[stepId]/route.ts b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/steps/[stepId]/route.ts index e929f9f..48b3300 100644 --- a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/steps/[stepId]/route.ts +++ b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/steps/[stepId]/route.ts @@ -10,7 +10,7 @@ type Params = { export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId, initiativeId, planId, stepId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const step = await storage.getPlanStep(boardId, initiativeId, planId, stepId); if (!step) { @@ -41,7 +41,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { if ("assigneeAgentId" in body) updates.assigneeAgentId = body.assigneeAgentId; if ("order" in body) updates.order = body.order; - const storage = getStorage(); + const storage = await getStorage(); const step = await storage.updatePlanStep(boardId, initiativeId, planId, stepId, updates); if (!step) { @@ -62,7 +62,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { export async function DELETE(_request: NextRequest, { params }: Params) { try { const { boardId, initiativeId, planId, stepId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const step = await storage.getPlanStep(boardId, initiativeId, planId, stepId); const deleted = await storage.deletePlanStep(boardId, initiativeId, planId, stepId); diff --git a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/steps/route.ts b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/steps/route.ts index 43923fa..e5c610a 100644 --- a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/steps/route.ts +++ b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/[planId]/steps/route.ts @@ -20,7 +20,7 @@ export async function POST(request: NextRequest, { params }: Params) { ); } - const storage = getStorage(); + const storage = await getStorage(); const plan = await storage.getPlan(boardId, initiativeId, planId); if (!plan) { return NextResponse.json( @@ -48,7 +48,7 @@ export async function POST(request: NextRequest, { params }: Params) { export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId, initiativeId, planId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const steps = await storage.listPlanSteps(boardId, initiativeId, planId); return NextResponse.json({ ok: true, data: steps }); } catch (err) { diff --git a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/route.ts b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/route.ts index 1a37d95..64928e5 100644 --- a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/route.ts +++ b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/plans/route.ts @@ -20,7 +20,7 @@ export async function POST(request: NextRequest, { params }: Params) { ); } - const storage = getStorage(); + const storage = await getStorage(); const initiative = await storage.getInitiative(boardId, initiativeId); if (!initiative) { return NextResponse.json( @@ -48,7 +48,7 @@ export async function POST(request: NextRequest, { params }: Params) { export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId, initiativeId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const plans = await storage.listPlans(boardId, initiativeId); return NextResponse.json({ ok: true, data: plans }); } catch (err) { diff --git a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/route.ts b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/route.ts index 4a0cf2f..b16e4f8 100644 --- a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/route.ts +++ b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/route.ts @@ -8,7 +8,7 @@ type Params = { params: Promise<{ boardId: string; initiativeId: string }> }; export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId, initiativeId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const initiative = await storage.getInitiative(boardId, initiativeId); if (!initiative) { @@ -39,7 +39,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { if ("kind" in body) updates.kind = body.kind; if ("assigneeAgentIds" in body) updates.assigneeAgentIds = body.assigneeAgentIds; - const storage = getStorage(); + const storage = await getStorage(); const initiative = await storage.updateInitiative(boardId, initiativeId, updates); if (!initiative) { @@ -62,7 +62,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { export async function DELETE(_request: NextRequest, { params }: Params) { try { const { boardId, initiativeId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const initiative = await storage.getInitiative(boardId, initiativeId); const deleted = await storage.deleteInitiative(boardId, initiativeId); diff --git a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/tasks/[taskId]/route.ts b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/tasks/[taskId]/route.ts index 393b899..7d04990 100644 --- a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/tasks/[taskId]/route.ts +++ b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/tasks/[taskId]/route.ts @@ -9,7 +9,7 @@ type Params = { params: Promise<{ boardId: string; initiativeId: string; taskId: export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId, initiativeId, taskId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const task = await storage.getTask(boardId, initiativeId, taskId); if (!task) { @@ -39,7 +39,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { { status: 400 }, ); } - const storage = getStorage(); + const storage = await getStorage(); const actor = await storage.getAgent(boardId, actorAgentId); if (!actor) { return NextResponse.json( @@ -96,7 +96,7 @@ export async function DELETE(_request: NextRequest, { params }: Params) { } const { boardId, initiativeId, taskId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const actor = await storage.getAgent(boardId, actorAgentId); if (!actor) { return NextResponse.json( diff --git a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/tasks/route.ts b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/tasks/route.ts index c81ddd3..6459e87 100644 --- a/src/app/api/boards/[boardId]/initiatives/[initiativeId]/tasks/route.ts +++ b/src/app/api/boards/[boardId]/initiatives/[initiativeId]/tasks/route.ts @@ -27,7 +27,7 @@ export async function POST(request: NextRequest, { params }: Params) { ); } - const storage = getStorage(); + const storage = await getStorage(); const actor = await storage.getAgent(boardId, actorAgentId); if (!actor) { return NextResponse.json( @@ -68,7 +68,7 @@ export async function GET(request: NextRequest, { params }: Params) { const { boardId, initiativeId } = await params; const url = new URL(request.url); - const storage = getStorage(); + const storage = await getStorage(); const tasks = await storage.listTasks(boardId, initiativeId, { status: url.searchParams.get("status") || undefined, assignee: url.searchParams.get("assignee") || undefined, diff --git a/src/app/api/boards/[boardId]/initiatives/route.ts b/src/app/api/boards/[boardId]/initiatives/route.ts index 78bbb8d..aeb73e0 100644 --- a/src/app/api/boards/[boardId]/initiatives/route.ts +++ b/src/app/api/boards/[boardId]/initiatives/route.ts @@ -20,7 +20,7 @@ export async function POST(request: NextRequest, { params }: Params) { ); } - const storage = getStorage(); + const storage = await getStorage(); const board = await storage.getBoard(boardId); if (!board) { return NextResponse.json( @@ -50,7 +50,7 @@ export async function POST(request: NextRequest, { params }: Params) { export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const initiatives = await storage.listInitiatives(boardId); return NextResponse.json({ ok: true, data: initiatives }); } catch (err) { diff --git a/src/app/api/boards/[boardId]/install/route.ts b/src/app/api/boards/[boardId]/install/route.ts index 75ece9d..dd27eeb 100644 --- a/src/app/api/boards/[boardId]/install/route.ts +++ b/src/app/api/boards/[boardId]/install/route.ts @@ -195,7 +195,7 @@ export async function GET(request: NextRequest, { params }: Params) { const agentName = url.searchParams.get("name") || agentType + "-agent"; const baseUrl = url.searchParams.get("url") || `${url.protocol}//${url.host}`; - const storage = getStorage(); + const storage = await getStorage(); const board = await storage.getBoard(boardId); if (!board) { return NextResponse.json( diff --git a/src/app/api/boards/[boardId]/projects/[projectId]/route.ts b/src/app/api/boards/[boardId]/projects/[projectId]/route.ts index ac1451e..f463124 100644 --- a/src/app/api/boards/[boardId]/projects/[projectId]/route.ts +++ b/src/app/api/boards/[boardId]/projects/[projectId]/route.ts @@ -8,7 +8,7 @@ type Params = { params: Promise<{ boardId: string; projectId: string }> }; export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId, projectId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const project = await storage.getProject(boardId, projectId); if (!project) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Project not found" } }, { status: 404 }); @@ -34,7 +34,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { if ("kind" in body) updates.kind = body.kind; if ("assigneeAgentIds" in body) updates.assigneeAgentIds = body.assigneeAgentIds; - const storage = getStorage(); + const storage = await getStorage(); const project = await storage.updateProject(boardId, projectId, updates); if (!project) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Project not found" } }, { status: 404 }); @@ -51,7 +51,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { export async function DELETE(_request: NextRequest, { params }: Params) { try { const { boardId, projectId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const project = await storage.getProject(boardId, projectId); const deleted = await storage.deleteProject(boardId, projectId); if (!deleted) { diff --git a/src/app/api/boards/[boardId]/projects/[projectId]/tasks/[taskId]/route.ts b/src/app/api/boards/[boardId]/projects/[projectId]/tasks/[taskId]/route.ts index 8467475..e0209c3 100644 --- a/src/app/api/boards/[boardId]/projects/[projectId]/tasks/[taskId]/route.ts +++ b/src/app/api/boards/[boardId]/projects/[projectId]/tasks/[taskId]/route.ts @@ -9,7 +9,7 @@ type Params = { params: Promise<{ boardId: string; projectId: string; taskId: st export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId, projectId, taskId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const task = await storage.getTask(boardId, projectId, taskId); if (!task) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Task not found" } }, { status: 404 }); @@ -34,7 +34,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { { status: 400 }, ); } - const storage = getStorage(); + const storage = await getStorage(); const actor = await storage.getAgent(boardId, actorAgentId); if (!actor) { return NextResponse.json( @@ -86,7 +86,7 @@ export async function DELETE(_request: NextRequest, { params }: Params) { } const { boardId, projectId, taskId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const actor = await storage.getAgent(boardId, actorAgentId); if (!actor) { return NextResponse.json( diff --git a/src/app/api/boards/[boardId]/projects/[projectId]/tasks/route.ts b/src/app/api/boards/[boardId]/projects/[projectId]/tasks/route.ts index 1335cf6..12b3b8a 100644 --- a/src/app/api/boards/[boardId]/projects/[projectId]/tasks/route.ts +++ b/src/app/api/boards/[boardId]/projects/[projectId]/tasks/route.ts @@ -19,7 +19,7 @@ export async function POST(request: NextRequest, { params }: Params) { if (!actorAgentId) { return NextResponse.json({ ok: false, error: { code: "MISSING_ACTOR", message: "actorAgentId is required" } }, { status: 400 }); } - const storage = getStorage(); + const storage = await getStorage(); const actor = await storage.getAgent(boardId, actorAgentId); if (!actor) { return NextResponse.json( @@ -59,7 +59,7 @@ export async function GET(request: NextRequest, { params }: Params) { assignee: url.searchParams.get("assignee") || undefined, tag: url.searchParams.get("tag") || undefined, }; - const storage = getStorage(); + const storage = await getStorage(); const tasks = await storage.listTasks(boardId, projectId, filters); return NextResponse.json({ ok: true, data: tasks }); } catch (err) { diff --git a/src/app/api/boards/[boardId]/projects/route.ts b/src/app/api/boards/[boardId]/projects/route.ts index b96dfdf..b0a55cf 100644 --- a/src/app/api/boards/[boardId]/projects/route.ts +++ b/src/app/api/boards/[boardId]/projects/route.ts @@ -14,7 +14,7 @@ export async function POST(request: NextRequest, { params }: Params) { if (!body.name) { return NextResponse.json({ ok: false, error: { code: "MISSING_NAME", message: "Project name is required" } }, { status: 400 }); } - const storage = getStorage(); + const storage = await getStorage(); const board = await storage.getBoard(boardId); if (!board) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Board not found" } }, { status: 404 }); @@ -35,7 +35,7 @@ export async function POST(request: NextRequest, { params }: Params) { export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const projects = await storage.listProjects(boardId); return NextResponse.json({ ok: true, data: projects }); } catch (err) { diff --git a/src/app/api/boards/[boardId]/route.ts b/src/app/api/boards/[boardId]/route.ts index 4b12c96..a82256a 100644 --- a/src/app/api/boards/[boardId]/route.ts +++ b/src/app/api/boards/[boardId]/route.ts @@ -8,7 +8,7 @@ type Params = { params: Promise<{ boardId: string }> }; export async function GET(_request: NextRequest, { params }: Params) { try { const { boardId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const summary = await storage.getBoardSummary(boardId); if (!summary) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Board not found" } }, { status: 404 }); @@ -31,7 +31,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { if ("name" in body) updates.name = body.name; if ("description" in body) updates.description = body.description; - const storage = getStorage(); + const storage = await getStorage(); const board = await storage.updateBoard(boardId, updates); if (!board) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Board not found" } }, { status: 404 }); @@ -47,7 +47,7 @@ export async function PATCH(request: NextRequest, { params }: Params) { export async function DELETE(_request: NextRequest, { params }: Params) { try { const { boardId } = await params; - const storage = getStorage(); + const storage = await getStorage(); const deleted = await storage.deleteBoard(boardId); if (!deleted) { return NextResponse.json({ ok: false, error: { code: "NOT_FOUND", message: "Board not found" } }, { status: 404 }); diff --git a/src/app/api/boards/[boardId]/tasks/route.ts b/src/app/api/boards/[boardId]/tasks/route.ts index e7e95e8..5e5fe94 100644 --- a/src/app/api/boards/[boardId]/tasks/route.ts +++ b/src/app/api/boards/[boardId]/tasks/route.ts @@ -13,7 +13,7 @@ export async function GET(request: NextRequest, { params }: Params) { assignee: url.searchParams.get("assignee") || undefined, tag: url.searchParams.get("tag") || undefined, }; - const storage = getStorage(); + const storage = await getStorage(); const tasks = await storage.listAllBoardTasks(boardId, filters); return NextResponse.json({ ok: true, data: tasks }); } catch (err) { diff --git a/src/app/api/boards/[boardId]/webhook/route.ts b/src/app/api/boards/[boardId]/webhook/route.ts index 3edfcef..0d1e5f6 100644 --- a/src/app/api/boards/[boardId]/webhook/route.ts +++ b/src/app/api/boards/[boardId]/webhook/route.ts @@ -44,7 +44,7 @@ export async function POST(request: NextRequest, { params }: Params) { ); } - const storage = getStorage(); + const storage = await getStorage(); const board = await storage.getBoard(boardId); if (!board) { return NextResponse.json( diff --git a/src/app/api/boards/route.ts b/src/app/api/boards/route.ts index 306da57..db9681a 100644 --- a/src/app/api/boards/route.ts +++ b/src/app/api/boards/route.ts @@ -10,7 +10,7 @@ export async function POST(request: NextRequest) { if (!body.name) { return NextResponse.json({ ok: false, error: { code: "MISSING_NAME", message: "Board name is required" } }, { status: 400 }); } - const storage = getStorage(); + const storage = await getStorage(); const board = await storage.createBoard({ name: body.name as string, description: (body.description as string) || "" }); return NextResponse.json({ ok: true, data: board }, { status: 201 }); } catch (err) { @@ -21,7 +21,7 @@ export async function POST(request: NextRequest) { export async function GET() { try { - const storage = getStorage(); + const storage = await getStorage(); const boards = await storage.listBoards(); return NextResponse.json({ ok: true, data: boards }); } catch (err) { diff --git a/src/app/skill.md/route.ts b/src/app/skill.md/route.ts index 3347fed..24e621b 100644 --- a/src/app/skill.md/route.ts +++ b/src/app/skill.md/route.ts @@ -15,7 +15,7 @@ export async function GET(request: NextRequest) { let projectsTable = ""; try { - const storage = getStorage(); + const storage = await getStorage(); const projects = await storage.listProjects(boardId); if (projects.length > 0) { projectsTable = projects diff --git a/src/lib/storage/blob-object-store.test.ts b/src/lib/storage/blob-object-store.test.ts new file mode 100644 index 0000000..c2d3edf --- /dev/null +++ b/src/lib/storage/blob-object-store.test.ts @@ -0,0 +1,180 @@ +import { beforeEach, describe, expect, mock, test } from "bun:test"; +import type { Agent, Initiative } from "@/lib/types"; + +// In-memory fake of the @vercel/blob SDK surface that BlobObjectStore uses. It +// faithfully models the behaviors the code depends on: not-found semantics, +// conditional create (allowOverwrite:false), and folded prefix listing. This +// lets us validate BlobObjectStore's key/prefix translation end-to-end without a +// live Blob store. +const blobs = new Map(); + +class FakeBlobNotFoundError extends Error { + constructor() { + super("Blob not found"); + this.name = "BlobNotFoundError"; + } +} + +function toBlob(pathname: string) { + return { + pathname, + url: `https://fake.blob/${pathname}`, + downloadUrl: `https://fake.blob/${pathname}?download=1`, + size: (blobs.get(pathname) ?? "").length, + uploadedAt: new Date(), + etag: "etag", + }; +} + +mock.module("@vercel/blob", () => ({ + BlobNotFoundError: FakeBlobNotFoundError, + async put(pathname: string, body: string, opts: { allowOverwrite?: boolean }) { + if (opts?.allowOverwrite === false && blobs.has(pathname)) { + throw new Error(`blob already exists: ${pathname}`); + } + blobs.set(pathname, typeof body === "string" ? body : String(body)); + return toBlob(pathname); + }, + async get(pathname: string) { + if (!blobs.has(pathname)) return null; + return { statusCode: 200, stream: new Response(blobs.get(pathname)).body, headers: new Headers(), blob: toBlob(pathname) }; + }, + async head(pathname: string) { + if (!blobs.has(pathname)) throw new FakeBlobNotFoundError(); + return toBlob(pathname); + }, + async del(target: string | string[]) { + for (const t of Array.isArray(target) ? target : [target]) blobs.delete(t); + }, + async copy(from: string, to: string) { + blobs.set(to, blobs.get(from) ?? ""); + return toBlob(to); + }, + async list(opts: { prefix?: string; mode?: "expanded" | "folded" } = {}) { + const prefix = opts.prefix ?? ""; + const keys = [...blobs.keys()].filter((k) => k.startsWith(prefix)); + if (opts.mode === "folded") { + const direct: string[] = []; + const folders = new Set(); + for (const k of keys) { + const rest = k.slice(prefix.length); + if (rest.includes("/")) folders.add(`${prefix}${rest.split("/")[0]}/`); + else direct.push(k); + } + return { blobs: direct.map(toBlob), folders: [...folders], hasMore: false, cursor: undefined }; + } + return { blobs: keys.map(toBlob), hasMore: false, cursor: undefined }; + }, +})); + +const { BlobObjectStore } = await import("./blob-object-store"); +const { DocumentStorage } = await import("./document-storage"); + +let storage: InstanceType; + +beforeEach(() => { + blobs.clear(); + storage = new DocumentStorage(new BlobObjectStore()); +}); + +function intro(sessionKey: string) { + return { intro: { runtime: "claude-code", sessionKey, thread: { id: sessionKey, name: sessionKey }, workingDirectory: "/tmp" } }; +} + +async function seed(): Promise<{ boardId: string; agent: Agent; initiative: Initiative }> { + const board = await storage.createBoard({ name: "Board", description: "" }); + const agent = await storage.createAgent(board.id, { name: "Worker", description: "", metadata: intro("s1") }); + const initiative = await storage.createInitiative(board.id, { name: "Init", description: "" }); + return { boardId: board.id, agent, initiative }; +} + +describe("BlobObjectStore via DocumentStorage", () => { + test("create/list/get round-trip through blob keys", async () => { + const { boardId, agent, initiative } = await seed(); + const task = await storage.createTask( + boardId, + initiative.id, + { title: "Do thing", description: "", assigneeAgentId: agent.id, priority: "medium", tags: [] }, + agent.id, + ); + + expect(await storage.getTask(boardId, initiative.id, task.id)).not.toBeNull(); + expect((await storage.listTasks(boardId, initiative.id)).length).toBe(1); + expect((await storage.listInitiatives(boardId)).map((i) => i.id)).toContain(initiative.id); + expect((await storage.listAgents(boardId)).length).toBe(1); + // Deterministic pathname (no random suffix) at the expected key. + expect(blobs.has(`boards/${boardId}/initiatives/${initiative.id}/tasks/${task.id}.json`)).toBe(true); + }); + + test("listChildren folding: initiatives dir yields only the .json entities", async () => { + const { boardId, initiative } = await seed(); + // The initiative file and its subtree (tasks/plans) coexist under initiatives/. + const initiatives = await storage.listInitiatives(boardId); + expect(initiatives.map((i) => i.id)).toEqual([initiative.id]); + }); + + test("unique-id reservation handles colliding slugs", async () => { + const { boardId, agent, initiative } = await seed(); + const results = await Promise.all( + Array.from({ length: 10 }, () => + storage.createTask( + boardId, + initiative.id, + { title: "Same", description: "", assigneeAgentId: null, priority: "medium", tags: [] }, + agent.id, + ), + ), + ); + expect(new Set(results.map((t) => t.id)).size).toBe(10); + }); + + test("deleteInitiative removes the file and the whole subtree", async () => { + const { boardId, agent, initiative } = await seed(); + await storage.createTask( + boardId, + initiative.id, + { title: "T", description: "", assigneeAgentId: null, priority: "medium", tags: [] }, + agent.id, + ); + expect(await storage.deleteInitiative(boardId, initiative.id)).toBe(true); + const leftover = [...blobs.keys()].filter((k) => k.includes(`/initiatives/${initiative.id}`)); + expect(leftover).toEqual([]); + }); + + test("activity events are stored and listed", async () => { + const { boardId, agent, initiative } = await seed(); + await storage.createTask( + boardId, + initiative.id, + { title: "T", description: "", assigneeAgentId: null, priority: "medium", tags: [] }, + agent.id, + ); + const activity = await storage.listActivity(boardId); + expect(activity.map((e) => e.type)).toContain("task.created"); + }); + + test("corrupt blob is quarantined and skipped on list", async () => { + const { boardId, agent, initiative } = await seed(); + const bad = await storage.createTask( + boardId, + initiative.id, + { title: "Bad", description: "", assigneeAgentId: null, priority: "medium", tags: [] }, + agent.id, + ); + const key = `boards/${boardId}/initiatives/${initiative.id}/tasks/${bad.id}.json`; + blobs.set(key, "{ not valid json"); + + const tasks = await storage.listTasks(boardId, initiative.id); + expect(tasks.map((t) => t.id)).not.toContain(bad.id); + // Moved out of the tasks dir into the quarantine prefix. + expect(blobs.has(key)).toBe(false); + expect([...blobs.keys()].some((k) => k.startsWith(".quarantine/"))).toBe(true); + }); + + test("getBoardSummary counts via blob backend", async () => { + const { boardId } = await seed(); + const summary = await storage.getBoardSummary(boardId); + expect(summary?.agentCount).toBe(1); + expect(summary?.initiativeCount).toBe(1); + }); +}); diff --git a/src/lib/storage/blob-object-store.ts b/src/lib/storage/blob-object-store.ts new file mode 100644 index 0000000..eb927c1 --- /dev/null +++ b/src/lib/storage/blob-object-store.ts @@ -0,0 +1,214 @@ +import { BlobNotFoundError, copy, del, get, head, list, put } from "@vercel/blob"; +import { randomUUID } from "crypto"; +import { timestamp } from "@/lib/utils"; +import type { ObjectStore } from "./object-store"; + +const ACCESS = "private" as const; +const QUARANTINE_PREFIX = ".quarantine"; + +const appendQueues = new Map>(); + +function withAppendLock(key: string, fn: () => Promise): Promise { + const previous = appendQueues.get(key) ?? Promise.resolve(); + const run = previous.then(fn, fn); + const settled = run.then( + () => undefined, + () => undefined, + ); + appendQueues.set(key, settled); + void settled.then(() => { + if (appendQueues.get(key) === settled) { + appendQueues.delete(key); + } + }); + return run; +} + +function isNotFound(err: unknown): boolean { + return err instanceof BlobNotFoundError; +} + +/** + * Vercel Blob-backed ObjectStore for production / serverless deployments, where the + * filesystem is read-only. Keys map directly to blob pathnames. + * + * Consistency notes: + * - Reads use `useCache: false` to avoid stale CDN responses after a write. + * - `createIfAbsent` relies on `allowOverwrite: false` for atomic reservation. + * - There is no cross-instance lock, so concurrent writers to the *same* key + * are last-write-wins (each write is itself atomic). Unique-id reservation + * prevents the common create collision; updates to one record assume a single + * logical writer, which matches the app's per-entity access pattern. + */ +export class BlobObjectStore implements ObjectStore { + private async readText(key: string): Promise { + let res; + try { + res = await get(key, { access: ACCESS, useCache: false }); + } catch (err) { + if (isNotFound(err)) return null; + throw err; + } + if (!res || res.statusCode !== 200 || !res.stream) return null; + return await new Response(res.stream).text(); + } + + private async quarantine(key: string, reason: string): Promise { + try { + const dest = `${QUARANTINE_PREFIX}/${timestamp().replace(/[:.]/g, "-")}-${randomUUID().slice(0, 8)}-${key.replace(/\//g, "_")}`; + // Only delete the source after the copy succeeds — matching FsObjectStore, + // where a failed rename leaves the original intact. If copy throws we fall + // to the catch below and the corrupt blob is preserved for inspection. + await copy(key, dest, { access: ACCESS }); + await del(key); + console.error(`Quarantined corrupt blob ${key} -> ${dest} (${reason})`); + } catch (err) { + console.error(`Failed to quarantine blob ${key}:`, err); + } + } + + async get(key: string, validate?: (value: unknown) => boolean): Promise { + const content = await this.readText(key); + if (content === null) return null; + + let parsed: unknown; + try { + parsed = JSON.parse(content); + } catch (err) { + console.error(`Corrupted JSON in blob ${key}:`, (err as Error).message); + await this.quarantine(key, "invalid JSON"); + return null; + } + + if (validate && !validate(parsed)) { + console.error(`Schema validation failed for blob ${key}; skipping`); + return null; + } + + return parsed as T; + } + + async put(key: string, data: unknown): Promise { + await put(key, JSON.stringify(data, null, 2), { + access: ACCESS, + addRandomSuffix: false, + allowOverwrite: true, + contentType: "application/json", + }); + } + + async createIfAbsent(key: string): Promise { + try { + await put(key, "{}", { + access: ACCESS, + addRandomSuffix: false, + allowOverwrite: false, + contentType: "application/json", + }); + return true; + } catch (err) { + // allowOverwrite:false rejects when the key exists. Disambiguate a real + // conflict from a transient error by confirming the blob is present. + if (await this.exists(key)) return false; + throw err; + } + } + + async exists(key: string): Promise { + try { + // head() takes no `access` option (its type is BlobCommandOptions) — it is + // token-scoped and resolves private blobs via BLOB_READ_WRITE_TOKEN. + // Passing `access` here is a type error and breaks the build. + await head(key); + return true; + } catch (err) { + if (isNotFound(err)) return false; + throw err; + } + } + + async delete(key: string): Promise { + try { + await del(key); + } catch (err) { + if (!isNotFound(err)) throw err; + } + } + + private async listPathnames(prefix: string): Promise { + const pathnames: string[] = []; + let cursor: string | undefined; + do { + const res = await list({ prefix, cursor, limit: 1000 }); + for (const blob of res.blobs) pathnames.push(blob.pathname); + cursor = res.hasMore ? res.cursor : undefined; + } while (cursor); + return pathnames; + } + + async deletePrefix(prefix: string): Promise { + const p = prefix.endsWith("/") ? prefix : `${prefix}/`; + const pathnames = await this.listPathnames(p); + // del accepts up to 1000 urls/pathnames per call. + for (let i = 0; i < pathnames.length; i += 1000) { + await del(pathnames.slice(i, i + 1000)); + } + } + + async listChildren(prefix: string): Promise { + const p = prefix.endsWith("/") ? prefix : `${prefix}/`; + const names = new Set(); + let cursor: string | undefined; + do { + const res = await list({ prefix: p, mode: "folded", cursor, limit: 1000 }); + for (const blob of res.blobs) { + const rel = blob.pathname.slice(p.length); + if (rel && !rel.includes("/")) names.add(rel); + } + for (const folder of res.folders) { + const rel = folder.slice(p.length).replace(/\/$/, ""); + if (rel) names.add(rel); + } + cursor = res.hasMore ? res.cursor : undefined; + } while (cursor); + return [...names]; + } + + private eventFile(eventsKey: string, bucket: string): string { + return `${eventsKey}/${bucket}.ndjson`; + } + + async appendEvent(eventsKey: string, bucket: string, record: unknown): Promise { + const key = this.eventFile(eventsKey, bucket); + // Serialize appends to the same daily file so concurrent events cannot + // interleave partial lines — matching FsObjectStore. + await withAppendLock(key, async () => { + const existing = (await this.readText(key)) ?? ""; + await put(key, `${existing}${JSON.stringify(record)}\n`, { + access: ACCESS, + addRandomSuffix: false, + allowOverwrite: true, + contentType: "application/x-ndjson", + }); + }); + } + + async listEventBuckets(eventsKey: string): Promise { + const files = await this.listChildren(eventsKey); + return files.filter((name) => name.endsWith(".ndjson")).map((name) => name.slice(0, -".ndjson".length)); + } + + async readEventBucket(eventsKey: string, bucket: string): Promise { + const content = await this.readText(this.eventFile(eventsKey, bucket)); + if (!content) return []; + const records: unknown[] = []; + for (const line of content.split("\n").filter(Boolean)) { + try { + records.push(JSON.parse(line)); + } catch { + // Skip malformed lines and continue. + } + } + return records; + } +} diff --git a/src/lib/storage/document-storage.test.ts b/src/lib/storage/document-storage.test.ts new file mode 100644 index 0000000..07fa504 --- /dev/null +++ b/src/lib/storage/document-storage.test.ts @@ -0,0 +1,130 @@ +import { beforeEach, describe, expect, test } from "bun:test"; +import { DocumentStorage } from "./document-storage"; +import { InMemoryObjectStore } from "./memory-object-store"; +import type { Agent, Initiative } from "@/lib/types"; + +// Exercises the backend-agnostic business logic against an in-memory ObjectStore. +// The same DocumentStorage runs on FsObjectStore (local) and BlobObjectStore (prod), so +// these guarantees hold for every backend. +let storage: DocumentStorage; + +beforeEach(() => { + storage = new DocumentStorage(new InMemoryObjectStore()); +}); + +function intro(sessionKey: string) { + return { intro: { runtime: "claude-code", sessionKey, thread: { id: sessionKey, name: sessionKey }, workingDirectory: "/tmp" } }; +} + +async function seed(): Promise<{ boardId: string; agent: Agent; initiative: Initiative }> { + const board = await storage.createBoard({ name: "Board", description: "" }); + const agent = await storage.createAgent(board.id, { name: "Worker", description: "", metadata: intro("s1") }); + const initiative = await storage.createInitiative(board.id, { name: "Init", description: "" }); + return { boardId: board.id, agent, initiative }; +} + +describe("DocumentStorage core flow", () => { + test("create/list/get board, initiative, task round-trip", async () => { + const { boardId, agent, initiative } = await seed(); + const task = await storage.createTask( + boardId, + initiative.id, + { title: "Do thing", description: "", assigneeAgentId: agent.id, priority: "high", tags: ["x"] }, + agent.id, + ); + expect(task.assigneeAgentIds).toEqual([agent.id]); + + const fetched = await storage.getTask(boardId, initiative.id, task.id); + expect(fetched?.title).toBe("Do thing"); + + const boards = await storage.listBoards(); + expect(boards.map((b) => b.id)).toContain(boardId); + + const tasks = await storage.listTasks(boardId, initiative.id); + expect(tasks.length).toBe(1); + }); + + test("concurrent identical-title creates get unique ids", async () => { + const { boardId, agent, initiative } = await seed(); + const results = await Promise.all( + Array.from({ length: 20 }, () => + storage.createTask( + boardId, + initiative.id, + { title: "Same", description: "", assigneeAgentId: null, priority: "medium", tags: [] }, + agent.id, + ), + ), + ); + expect(new Set(results.map((t) => t.id)).size).toBe(20); + expect((await storage.listTasks(boardId, initiative.id)).length).toBe(20); + }); + + test("status change to in_progress wires up the agent pointer", async () => { + const { boardId, agent, initiative } = await seed(); + const task = await storage.createTask( + boardId, + initiative.id, + { title: "T", description: "", assigneeAgentId: agent.id, priority: "medium", tags: [] }, + agent.id, + ); + await storage.updateTask(boardId, initiative.id, task.id, { status: "in_progress" }, agent.id); + const after = await storage.getAgent(boardId, agent.id); + expect(after?.currentTaskId).toBe(task.id); + expect(after?.currentInitiativeId).toBe(initiative.id); + }); + + test("deleting an initiative removes its task subtree", async () => { + const { boardId, agent, initiative } = await seed(); + await storage.createTask( + boardId, + initiative.id, + { title: "T", description: "", assigneeAgentId: null, priority: "medium", tags: [] }, + agent.id, + ); + expect(await storage.deleteInitiative(boardId, initiative.id)).toBe(true); + expect(await storage.getInitiative(boardId, initiative.id)).toBeNull(); + expect(await storage.listTasks(boardId, initiative.id)).toEqual([]); + }); + + test("duplicate agent name/session are rejected", async () => { + const board = await storage.createBoard({ name: "B", description: "" }); + await storage.createAgent(board.id, { name: "Dup", description: "", metadata: intro("k1") }); + await expect( + storage.createAgent(board.id, { name: "Dup", description: "", metadata: intro("k2") }), + ).rejects.toThrow(/already exists/); + }); + + test("activity log records and lists events newest-first", async () => { + const { boardId, agent, initiative } = await seed(); + await storage.createTask( + boardId, + initiative.id, + { title: "T", description: "", assigneeAgentId: null, priority: "medium", tags: [] }, + agent.id, + ); + const activity = await storage.listActivity(boardId); + const types = activity.map((e) => e.type); + expect(types).toContain("task.created"); + expect(types).toContain("agent.registered"); + }); + + test("getBoardSummary counts entities and isolates a failing section", async () => { + const { boardId } = await seed(); + + const summary = await storage.getBoardSummary(boardId); + expect(summary?.agentCount).toBe(1); + expect(summary?.initiativeCount).toBe(1); + + class Broken extends DocumentStorage { + async listAllBoardTasks(): Promise { + throw new Error("boom"); + } + } + const broken = new Broken(new InMemoryObjectStore()); + const b = await broken.createBoard({ name: "B2", description: "" }); + const brokenSummary = await broken.getBoardSummary(b.id); + expect(brokenSummary).not.toBeNull(); + expect(brokenSummary!.taskCount).toBe(0); + }); +}); diff --git a/src/lib/storage/document-storage.ts b/src/lib/storage/document-storage.ts new file mode 100644 index 0000000..2a02668 --- /dev/null +++ b/src/lib/storage/document-storage.ts @@ -0,0 +1,1399 @@ +import { randomUUID } from "crypto"; +import type { + ActivityEvent, + Agent, + Board, + BoardSummary, + Initiative, + Plan, + PlanStep, + Project, + Task, + TaskStatus, +} from "@/lib/types"; +import type { Storage } from "./index"; +import type { ObjectStore } from "./object-store"; +import { slugify, timestamp } from "@/lib/utils"; + +type AgentIntro = { + runtime: string; + sessionKey: string; + model: string; + thread: { + id: string; + name: string; + source?: string; + }; + workingDirectory: string; + host?: { + hostname: string; + localIp: string; + }; +}; + +export class AgentRegistrationError extends Error { + code: string; + + constructor(code: string, message: string) { + super(message); + this.name = "AgentRegistrationError"; + this.code = code; + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function normalizeAgentName(name: string): string { + return name.trim(); +} + +function canonicalAgentName(name: string): string { + return normalizeAgentName(name).toLowerCase(); +} + +function normalizeText(value: unknown): string { + return typeof value === "string" ? value.trim() : ""; +} + +function inferModelFromRuntime(runtime: string): string { + const key = runtime.toLowerCase(); + if (key.includes("codex") || key.includes("openai")) return "codex"; + if (key.includes("claude") || key.includes("anthropic")) return "claude"; + if (key.includes("cursor")) return "cursor"; + return "unknown-model"; +} + +function extractThreadContext(intro: Record): AgentIntro["thread"] | null { + const threadRaw = intro.thread; + + if (typeof threadRaw === "string") { + const id = normalizeText(threadRaw); + const name = normalizeText(intro.threadName) || id; + if (!id) return null; + return { id, name }; + } + + if (!isRecord(threadRaw)) return null; + + const id = normalizeText(threadRaw.id) || normalizeText(intro.threadId); + const name = normalizeText(threadRaw.name) || normalizeText(intro.threadName) || id; + const source = normalizeText(threadRaw.source); + if (!id) return null; + return source ? { id, name, source } : { id, name }; +} + +function extractAgentIntro(metadata: unknown): AgentIntro | null { + if (!isRecord(metadata)) return null; + const intro = metadata.intro; + if (!isRecord(intro)) return null; + + const runtime = normalizeText(intro.runtime) || "unknown-runtime"; + const sessionKey = normalizeText(intro.sessionKey) || normalizeText(intro.instanceKey); + const model = normalizeText(intro.model) || inferModelFromRuntime(runtime); + const thread = + extractThreadContext(intro) || + (() => { + const fallbackId = normalizeText(intro.threadId) || "unknown-thread"; + const fallbackName = normalizeText(intro.threadName) || fallbackId; + return { id: fallbackId, name: fallbackName, source: runtime }; + })(); + const workingDirectory = normalizeText(intro.workingDirectory) || "unknown-working-directory"; + const hostRaw = isRecord(intro.host) ? intro.host : {}; + const hostHostname = normalizeText(hostRaw.hostname); + const hostLocalIp = normalizeText(hostRaw.localIp); + const host = + hostHostname || hostLocalIp + ? { + hostname: hostHostname || "unknown-host", + localIp: hostLocalIp || "unknown-local-ip", + } + : undefined; + + if (!sessionKey) return null; + return { runtime, sessionKey, model, thread, workingDirectory, host }; +} + +// A stored entity is only usable if it parsed to an object carrying a string id. +// Anything else is either a transient reservation placeholder or structurally +// unusable, and is skipped on read. +export function isEntityRecord(value: unknown): boolean { + return isRecord(value) && typeof (value as { id?: unknown }).id === "string" && (value as { id: string }).id.length > 0; +} + +// Defensive string compare for sort keys that may be missing on partially +// written or legacy records — `undefined.localeCompare` would throw. +function compareStr(a: string | undefined, b: string | undefined): number { + return (a ?? "").localeCompare(b ?? ""); +} + +function normalizeAssignees(task: Partial): string[] { + if (task.assigneeAgentIds && task.assigneeAgentIds.length > 0) { + return Array.from(new Set(task.assigneeAgentIds.filter(Boolean))); + } + if (task.assigneeAgentId) { + return [task.assigneeAgentId]; + } + return []; +} + +function normalizeTask(task: Task, fallbackInitiativeId?: string): Task { + const initiativeId = task.initiativeId || task.projectId || fallbackInitiativeId || "general"; + const assigneeAgentIds = normalizeAssignees(task); + const assigneeAgentId = task.assigneeAgentId || assigneeAgentIds[0] || null; + + return { + ...task, + initiativeId, + projectId: task.projectId || initiativeId, + planId: task.planId || null, + planStepId: task.planStepId || null, + assigneeAgentId, + assigneeAgentIds, + deliverables: task.deliverables || [], + }; +} + +// --- Key helpers (POSIX-style relative keys; identical layout across backends) --- + +function boardDir(boardId: string): string { + return `boards/${boardId}`; +} +function boardFile(boardId: string): string { + return `${boardDir(boardId)}/board.json`; +} +function agentsDir(boardId: string): string { + return `${boardDir(boardId)}/agents`; +} +function agentFile(boardId: string, agentId: string): string { + return `${agentsDir(boardId)}/${agentId}.json`; +} +function initiativesDir(boardId: string): string { + return `${boardDir(boardId)}/initiatives`; +} +function initiativeFile(boardId: string, initiativeId: string): string { + return `${initiativesDir(boardId)}/${initiativeId}.json`; +} +function initiativeSubdir(boardId: string, initiativeId: string): string { + return `${initiativesDir(boardId)}/${initiativeId}`; +} +function tasksDir(boardId: string, initiativeId: string): string { + return `${initiativeSubdir(boardId, initiativeId)}/tasks`; +} +function taskFile(boardId: string, initiativeId: string, taskId: string): string { + return `${tasksDir(boardId, initiativeId)}/${taskId}.json`; +} +function plansDir(boardId: string, initiativeId: string): string { + return `${initiativeSubdir(boardId, initiativeId)}/plans`; +} +function planFile(boardId: string, initiativeId: string, planId: string): string { + return `${plansDir(boardId, initiativeId)}/${planId}.json`; +} +function planSubdir(boardId: string, initiativeId: string, planId: string): string { + return `${plansDir(boardId, initiativeId)}/${planId}`; +} +function planStepsDir(boardId: string, initiativeId: string, planId: string): string { + return `${planSubdir(boardId, initiativeId, planId)}/steps`; +} +function planStepFile(boardId: string, initiativeId: string, planId: string, stepId: string): string { + return `${planStepsDir(boardId, initiativeId, planId)}/${stepId}.json`; +} +function eventsDir(boardId: string): string { + return `${boardDir(boardId)}/events`; +} +function legacyProjectsDir(boardId: string): string { + return `${boardDir(boardId)}/projects`; +} +function migrationMarkerFile(boardId: string): string { + return `${boardDir(boardId)}/.migration-v2.json`; +} + +/** + * Backend-agnostic storage implementation. All durability and concurrency + * primitives are delegated to the injected ObjectStore; this class owns the data + * model, validation, normalization, and activity logging. + */ +export class DocumentStorage implements Storage { + protected store: ObjectStore; + private initializedBoards = new Set(); + private initializingBoards = new Map>(); + + constructor(store: ObjectStore) { + this.store = store; + } + + private async ensureBoardStructure(boardId: string): Promise { + if (this.initializedBoards.has(boardId)) return; + const existingInit = this.initializingBoards.get(boardId); + if (existingInit) { + await existingInit; + return; + } + + const initPromise = (async () => { + await this.migrateLegacyProjects(boardId); + this.initializedBoards.add(boardId); + })(); + + this.initializingBoards.set(boardId, initPromise); + try { + await initPromise; + } finally { + this.initializingBoards.delete(boardId); + } + } + + private async migrateLegacyProjects(boardId: string): Promise { + if (await this.store.exists(migrationMarkerFile(boardId))) return; + + const projectIds = await this.store.listChildren(legacyProjectsDir(boardId)); + if (projectIds.length === 0) { + await this.store.put(migrationMarkerFile(boardId), { version: 2, migratedAt: timestamp(), migratedProjects: 0 }); + return; + } + + let migratedProjects = 0; + let migratedTasks = 0; + + for (const projectId of projectIds) { + const legacyProjectKey = `${legacyProjectsDir(boardId)}/${projectId}/project.json`; + const legacyProject = await this.store.get(legacyProjectKey, isEntityRecord); + if (!legacyProject) continue; + + const nextInitiative: Initiative = { + id: legacyProject.id, + boardId, + name: legacyProject.name, + description: legacyProject.description || "", + status: "active", + kind: "general", + assigneeAgentIds: [], + createdAt: legacyProject.createdAt || timestamp(), + updatedAt: legacyProject.updatedAt || legacyProject.createdAt || timestamp(), + }; + + if (!(await this.store.exists(initiativeFile(boardId, projectId)))) { + await this.store.put(initiativeFile(boardId, projectId), nextInitiative); + } + + const legacyTaskDir = `${legacyProjectsDir(boardId)}/${projectId}/tasks`; + const taskFiles = await this.store.listChildren(legacyTaskDir); + + for (const taskName of taskFiles) { + if (!taskName.endsWith(".json")) continue; + const legacyTask = await this.store.get(`${legacyTaskDir}/${taskName}`, isEntityRecord); + if (!legacyTask) continue; + + const normalized = normalizeTask(legacyTask, projectId); + const nextTaskKey = taskFile(boardId, projectId, normalized.id); + if (!(await this.store.exists(nextTaskKey))) { + await this.store.put(nextTaskKey, normalized); + migratedTasks += 1; + } + } + + migratedProjects += 1; + } + + await this.store.put(migrationMarkerFile(boardId), { + version: 2, + migratedAt: timestamp(), + migratedProjects, + migratedTasks, + }); + + await this.appendActivity(boardId, { + type: "system.migration.v2", + message: `Migrated ${migratedProjects} legacy project(s) and ${migratedTasks} task(s) to initiatives`, + payload: { migratedProjects, migratedTasks }, + }); + } + + private async resolveUniqueId(keyForId: (id: string) => string, desiredId: string): Promise { + let candidate = desiredId; + let counter = 2; + for (;;) { + if (await this.store.createIfAbsent(keyForId(candidate))) { + return candidate; + } + candidate = `${desiredId}-${counter}`; + counter += 1; + } + } + + private async appendActivity( + boardId: string, + event: Omit, + ): Promise { + const createdAt = timestamp(); + const activity: ActivityEvent = { + id: randomUUID(), + boardId, + createdAt, + ...event, + }; + + await this.store.appendEvent(eventsDir(boardId), createdAt.slice(0, 10), activity); + + return activity; + } + + private async writeAgent(boardId: string, agent: Agent): Promise { + await this.store.put(agentFile(boardId, agent.id), agent); + } + + private async requireRegisteredAgent(boardId: string, agentId: string): Promise { + const agent = await this.getAgent(boardId, agentId); + if (!agent) { + throw new Error(`Agent not registered: ${agentId}`); + } + return agent; + } + + private async syncAgentPointersForTask(boardId: string, task: Task): Promise { + const assignees = normalizeAssignees(task); + + if (task.status === "in_progress") { + for (const agentId of assignees) { + const agent = await this.getAgent(boardId, agentId); + if (!agent) continue; + const updated: Agent = { + ...agent, + currentTaskId: task.id, + currentInitiativeId: task.initiativeId, + lastHeartbeat: timestamp(), + }; + await this.writeAgent(boardId, updated); + } + return; + } + + const agents = await this.listAgents(boardId); + for (const agent of agents) { + if (agent.currentTaskId !== task.id || agent.currentInitiativeId !== task.initiativeId) { + continue; + } + const cleared: Agent = { + ...agent, + currentTaskId: null, + currentInitiativeId: null, + lastHeartbeat: timestamp(), + }; + await this.writeAgent(boardId, cleared); + } + } + + // --- Boards --- + + async createBoard(data: Pick): Promise { + const id = slugify(data.name); + const now = timestamp(); + const board: Board = { + id, + name: data.name, + description: data.description || "", + createdAt: now, + updatedAt: now, + }; + + await this.store.put(boardFile(id), board); + + return board; + } + + async listBoards(): Promise { + const dirs = await this.store.listChildren("boards"); + const boards: Board[] = []; + + for (const d of dirs) { + const board = await this.store.get(`${boardDir(d)}/board.json`, isEntityRecord); + if (board) boards.push(board); + } + + return boards.sort((a, b) => compareStr(b.createdAt, a.createdAt)); + } + + async getBoard(boardId: string): Promise { + return this.store.get(boardFile(boardId), isEntityRecord); + } + + async getBoardSummary(boardId: string): Promise { + await this.ensureBoardStructure(boardId); + + const board = await this.getBoard(boardId); + if (!board) return null; + + // Isolate each section: a failure in one count must not blank out the + // whole summary. Each section falls back to an empty result. + const safeSection = async (label: string, fn: () => Promise, fallback: T): Promise => { + try { + return await fn(); + } catch (err) { + console.error(`Board summary section "${label}" failed for ${boardId}:`, err); + return fallback; + } + }; + + const [agents, initiatives, tasks] = await Promise.all([ + safeSection("agents", () => this.listAgents(boardId), [] as Agent[]), + safeSection("initiatives", () => this.listInitiatives(boardId), [] as Initiative[]), + safeSection("tasks", () => this.listAllBoardTasks(boardId), [] as Task[]), + ]); + + const plansByInitiative = await Promise.all( + initiatives.map((initiative) => + safeSection(`plans:${initiative.id}`, () => this.listPlans(boardId, initiative.id), [] as Plan[]), + ), + ); + + const allPlans = plansByInitiative.flat(); + + const stepsByPlan = await Promise.all( + allPlans.map((plan) => + safeSection( + `steps:${plan.id}`, + () => this.listPlanSteps(boardId, plan.initiativeId, plan.id), + [] as PlanStep[], + ), + ), + ); + + const planStepCount = stepsByPlan.reduce((sum, steps) => sum + steps.length, 0); + + return { + ...board, + agentCount: agents.length, + initiativeCount: initiatives.length, + planCount: allPlans.length, + planStepCount, + projectCount: initiatives.length, + taskCount: tasks.length, + }; + } + + async updateBoard(boardId: string, data: Partial>): Promise { + await this.ensureBoardStructure(boardId); + + const board = await this.getBoard(boardId); + if (!board) return null; + + const updated = { ...board, ...data, updatedAt: timestamp() }; + await this.store.put(boardFile(boardId), updated); + + await this.appendActivity(boardId, { + type: "board.updated", + message: "Board details updated", + payload: { fields: Object.keys(data) }, + }); + + return updated; + } + + async deleteBoard(boardId: string): Promise { + if (!(await this.store.exists(boardFile(boardId)))) return false; + await this.store.deletePrefix(boardDir(boardId)); + this.initializedBoards.delete(boardId); + return true; + } + + // --- Agents --- + + async createAgent(boardId: string, data: Pick): Promise { + await this.ensureBoardStructure(boardId); + + const name = normalizeAgentName(data.name || ""); + if (!name) { + throw new AgentRegistrationError("MISSING_NAME", "Agent name is required"); + } + + const id = slugify(name); + if (!id) { + throw new AgentRegistrationError("INVALID_NAME", "Agent name must contain at least one letter or number"); + } + + const metadata = isRecord(data.metadata) ? data.metadata : {}; + const intro = extractAgentIntro(metadata); + if (!intro) { + throw new AgentRegistrationError( + "INVALID_AGENT_INTRO", + "Agent intro metadata requires sessionKey (runtime/model/thread/workingDirectory are auto-detected or defaulted)", + ); + } + + const existingById = await this.getAgent(boardId, id); + if (existingById) { + throw new AgentRegistrationError("AGENT_ID_CONFLICT", `Agent id already exists: ${id}`); + } + + const existingAgents = await this.listAgents(boardId); + const desiredName = canonicalAgentName(name); + for (const existing of existingAgents) { + if (canonicalAgentName(existing.name) === desiredName) { + throw new AgentRegistrationError("AGENT_NAME_CONFLICT", `Agent name already exists: ${name}`); + } + const existingIntro = extractAgentIntro(existing.metadata); + if (existingIntro?.sessionKey === intro.sessionKey) { + throw new AgentRegistrationError("AGENT_KEY_CONFLICT", `Agent key already exists: ${intro.sessionKey}`); + } + } + + const now = timestamp(); + + const agent: Agent = { + id, + boardId, + name, + description: data.description || "", + status: "idle", + statusMessage: "", + currentTaskId: null, + currentInitiativeId: null, + metadata, + registeredAt: now, + lastHeartbeat: now, + }; + + await this.writeAgent(boardId, agent); + + await this.appendActivity(boardId, { + type: "agent.registered", + message: `Agent ${agent.name} registered`, + agentId: agent.id, + }); + + return agent; + } + + async listAgents(boardId: string): Promise { + await this.ensureBoardStructure(boardId); + + const files = await this.store.listChildren(agentsDir(boardId)); + const agents: Agent[] = []; + + for (const f of files) { + if (!f.endsWith(".json")) continue; + const agent = await this.store.get(`${agentsDir(boardId)}/${f}`, isEntityRecord); + if (!agent) continue; + + agents.push({ + ...agent, + currentInitiativeId: agent.currentInitiativeId || null, + }); + } + + return agents.sort((a, b) => compareStr(a.name, b.name)); + } + + async getAgent(boardId: string, agentId: string): Promise { + await this.ensureBoardStructure(boardId); + + const agent = await this.store.get(agentFile(boardId, agentId), isEntityRecord); + if (!agent) return null; + + return { + ...agent, + currentInitiativeId: agent.currentInitiativeId || null, + }; + } + + async updateAgent( + boardId: string, + agentId: string, + data: Partial>, + ): Promise { + await this.ensureBoardStructure(boardId); + + const agent = await this.getAgent(boardId, agentId); + if (!agent) return null; + + const updated: Agent = { + ...agent, + ...data, + lastHeartbeat: timestamp(), + }; + + await this.writeAgent(boardId, updated); + + await this.appendActivity(boardId, { + type: "agent.updated", + message: `Agent ${updated.name} updated status to ${updated.status}`, + agentId, + payload: { fields: Object.keys(data) }, + }); + + return updated; + } + + async deleteAgent(boardId: string, agentId: string): Promise { + await this.ensureBoardStructure(boardId); + + if (!(await this.store.exists(agentFile(boardId, agentId)))) return false; + await this.store.delete(agentFile(boardId, agentId)); + + await this.appendActivity(boardId, { + type: "agent.removed", + message: `Agent ${agentId} removed`, + agentId, + }); + + return true; + } + + async heartbeatAgent(boardId: string, agentId: string, message?: string): Promise { + await this.ensureBoardStructure(boardId); + + const agent = await this.getAgent(boardId, agentId); + if (!agent) return null; + + const updated: Agent = { + ...agent, + status: "active", + lastHeartbeat: timestamp(), + ...(message !== undefined ? { statusMessage: message } : {}), + }; + + await this.writeAgent(boardId, updated); + + await this.appendActivity(boardId, { + type: "agent.heartbeat", + message: message ? `Heartbeat: ${message}` : "Heartbeat", + agentId, + initiativeId: updated.currentInitiativeId || undefined, + taskId: updated.currentTaskId || undefined, + }); + + return updated; + } + + // --- Initiatives --- + + async createInitiative( + boardId: string, + data: Pick & Partial>, + ): Promise { + await this.ensureBoardStructure(boardId); + + const baseId = slugify(data.name); + const id = await this.resolveUniqueId((candidate) => initiativeFile(boardId, candidate), baseId); + const now = timestamp(); + + const initiative: Initiative = { + id, + boardId, + name: data.name, + description: data.description || "", + status: data.status || "active", + kind: data.kind || "thread", + assigneeAgentIds: data.assigneeAgentIds || [], + createdAt: now, + updatedAt: now, + }; + + await this.store.put(initiativeFile(boardId, id), initiative); + + await this.appendActivity(boardId, { + type: "initiative.created", + message: `Initiative ${initiative.name} created`, + initiativeId: initiative.id, + payload: { kind: initiative.kind }, + }); + + return initiative; + } + + async listInitiatives(boardId: string): Promise { + await this.ensureBoardStructure(boardId); + + const files = await this.store.listChildren(initiativesDir(boardId)); + const initiatives: Initiative[] = []; + + for (const f of files) { + if (!f.endsWith(".json")) continue; + const initiative = await this.store.get(`${initiativesDir(boardId)}/${f}`, isEntityRecord); + if (initiative) { + initiatives.push({ + ...initiative, + status: initiative.status || "active", + kind: initiative.kind || "thread", + assigneeAgentIds: initiative.assigneeAgentIds || [], + }); + } + } + + return initiatives.sort((a, b) => compareStr(a.createdAt, b.createdAt)); + } + + async getInitiative(boardId: string, initiativeId: string): Promise { + await this.ensureBoardStructure(boardId); + + const initiative = await this.store.get(initiativeFile(boardId, initiativeId), isEntityRecord); + if (!initiative) return null; + + return { + ...initiative, + status: initiative.status || "active", + kind: initiative.kind || "thread", + assigneeAgentIds: initiative.assigneeAgentIds || [], + }; + } + + async updateInitiative( + boardId: string, + initiativeId: string, + data: Partial>, + ): Promise { + await this.ensureBoardStructure(boardId); + + const initiative = await this.getInitiative(boardId, initiativeId); + if (!initiative) return null; + + const updated: Initiative = { + ...initiative, + ...data, + updatedAt: timestamp(), + }; + + await this.store.put(initiativeFile(boardId, initiativeId), updated); + + await this.appendActivity(boardId, { + type: "initiative.updated", + message: `Initiative ${updated.name} updated`, + initiativeId, + payload: { fields: Object.keys(data) }, + }); + + return updated; + } + + async deleteInitiative(boardId: string, initiativeId: string): Promise { + await this.ensureBoardStructure(boardId); + + const hasFile = await this.store.exists(initiativeFile(boardId, initiativeId)); + const children = await this.store.listChildren(initiativeSubdir(boardId, initiativeId)); + if (!hasFile && children.length === 0) return false; + + await this.store.delete(initiativeFile(boardId, initiativeId)); + await this.store.deletePrefix(initiativeSubdir(boardId, initiativeId)); + + await this.appendActivity(boardId, { + type: "initiative.removed", + message: `Initiative ${initiativeId} removed`, + initiativeId, + }); + + return true; + } + + // --- Plans --- + + async createPlan( + boardId: string, + initiativeId: string, + data: Pick & Partial>, + ): Promise { + await this.ensureBoardStructure(boardId); + + const initiative = await this.getInitiative(boardId, initiativeId); + if (!initiative) { + throw new Error(`Initiative not found: ${initiativeId}`); + } + + const baseId = slugify(data.title); + const id = await this.resolveUniqueId((candidate) => planFile(boardId, initiativeId, candidate), baseId); + const now = timestamp(); + + const plan: Plan = { + id, + boardId, + initiativeId, + title: data.title, + description: data.description || "", + status: data.status || "todo", + ownerAgentId: data.ownerAgentId || null, + tags: data.tags || [], + createdAt: now, + updatedAt: now, + completedAt: null, + }; + + await this.store.put(planFile(boardId, initiativeId, id), plan); + + await this.appendActivity(boardId, { + type: "plan.created", + message: `Plan ${plan.title} created`, + initiativeId, + payload: { planId: plan.id, status: plan.status }, + agentId: plan.ownerAgentId || undefined, + }); + + return plan; + } + + async listPlans(boardId: string, initiativeId: string): Promise { + await this.ensureBoardStructure(boardId); + + const files = await this.store.listChildren(plansDir(boardId, initiativeId)); + const plans: Plan[] = []; + for (const fileName of files) { + if (!fileName.endsWith(".json")) continue; + const plan = await this.store.get(`${plansDir(boardId, initiativeId)}/${fileName}`, isEntityRecord); + if (!plan) continue; + plans.push({ + ...plan, + status: plan.status || "todo", + ownerAgentId: plan.ownerAgentId || null, + tags: plan.tags || [], + completedAt: plan.completedAt || null, + }); + } + + return plans.sort((a, b) => compareStr(a.createdAt, b.createdAt)); + } + + async getPlan(boardId: string, initiativeId: string, planId: string): Promise { + await this.ensureBoardStructure(boardId); + + const plan = await this.store.get(planFile(boardId, initiativeId, planId), isEntityRecord); + if (!plan) return null; + + return { + ...plan, + status: plan.status || "todo", + ownerAgentId: plan.ownerAgentId || null, + tags: plan.tags || [], + completedAt: plan.completedAt || null, + }; + } + + async updatePlan( + boardId: string, + initiativeId: string, + planId: string, + data: Partial>, + ): Promise { + await this.ensureBoardStructure(boardId); + + const current = await this.getPlan(boardId, initiativeId, planId); + if (!current) return null; + + const now = timestamp(); + const nextStatus = data.status || current.status; + const completedAt = + nextStatus === "done" ? now : data.status && data.status !== "done" ? null : current.completedAt; + + const updated: Plan = { + ...current, + ...data, + status: nextStatus, + updatedAt: now, + completedAt, + ownerAgentId: data.ownerAgentId === undefined ? current.ownerAgentId : data.ownerAgentId, + tags: data.tags || current.tags, + }; + + await this.store.put(planFile(boardId, initiativeId, planId), updated); + + const statusChange = data.status && data.status !== current.status; + await this.appendActivity(boardId, { + type: statusChange ? "plan.status_changed" : "plan.updated", + message: statusChange + ? `Plan ${updated.title} moved ${current.status} -> ${updated.status}` + : `Plan ${updated.title} updated`, + initiativeId, + payload: { planId, status: updated.status, fields: Object.keys(data) }, + agentId: updated.ownerAgentId || undefined, + }); + + return updated; + } + + async deletePlan(boardId: string, initiativeId: string, planId: string): Promise { + await this.ensureBoardStructure(boardId); + + const hasFile = await this.store.exists(planFile(boardId, initiativeId, planId)); + const children = await this.store.listChildren(planSubdir(boardId, initiativeId, planId)); + if (!hasFile && children.length === 0) return false; + + const tasks = await this.listTasks(boardId, initiativeId); + for (const task of tasks) { + if (task.planId !== planId) continue; + const detached: Task = { ...task, planId: null, planStepId: null, updatedAt: timestamp() }; + await this.store.put(taskFile(boardId, initiativeId, task.id), detached); + } + + await this.store.delete(planFile(boardId, initiativeId, planId)); + await this.store.deletePrefix(planSubdir(boardId, initiativeId, planId)); + + await this.appendActivity(boardId, { + type: "plan.removed", + message: `Plan ${planId} removed`, + initiativeId, + payload: { planId }, + }); + + return true; + } + + // --- Plan steps --- + + async createPlanStep( + boardId: string, + initiativeId: string, + planId: string, + data: Pick & Partial>, + ): Promise { + await this.ensureBoardStructure(boardId); + + const plan = await this.getPlan(boardId, initiativeId, planId); + if (!plan) { + throw new Error(`Plan not found: ${planId}`); + } + + const baseId = slugify(data.title); + const id = await this.resolveUniqueId( + (candidate) => planStepFile(boardId, initiativeId, planId, candidate), + baseId, + ); + const now = timestamp(); + const existingSteps = await this.listPlanSteps(boardId, initiativeId, planId); + const order = data.order ?? existingSteps.length + 1; + + const step: PlanStep = { + id, + boardId, + initiativeId, + planId, + title: data.title, + description: data.description || "", + status: data.status || "todo", + assigneeAgentId: data.assigneeAgentId || null, + order, + createdAt: now, + updatedAt: now, + completedAt: null, + }; + + await this.store.put(planStepFile(boardId, initiativeId, planId, id), step); + + await this.appendActivity(boardId, { + type: "plan_step.created", + message: `Plan step ${step.title} created`, + initiativeId, + payload: { planId, stepId: step.id, status: step.status, order: step.order }, + agentId: step.assigneeAgentId || undefined, + }); + + return step; + } + + async listPlanSteps(boardId: string, initiativeId: string, planId: string): Promise { + await this.ensureBoardStructure(boardId); + + const files = await this.store.listChildren(planStepsDir(boardId, initiativeId, planId)); + const steps: PlanStep[] = []; + for (const fileName of files) { + if (!fileName.endsWith(".json")) continue; + const step = await this.store.get( + `${planStepsDir(boardId, initiativeId, planId)}/${fileName}`, + isEntityRecord, + ); + if (!step) continue; + steps.push({ + ...step, + status: step.status || "todo", + assigneeAgentId: step.assigneeAgentId || null, + completedAt: step.completedAt || null, + }); + } + + return steps.sort((a, b) => (a.order ?? 0) - (b.order ?? 0) || compareStr(a.createdAt, b.createdAt)); + } + + async getPlanStep( + boardId: string, + initiativeId: string, + planId: string, + stepId: string, + ): Promise { + await this.ensureBoardStructure(boardId); + + const step = await this.store.get(planStepFile(boardId, initiativeId, planId, stepId), isEntityRecord); + if (!step) return null; + + return { + ...step, + status: step.status || "todo", + assigneeAgentId: step.assigneeAgentId || null, + completedAt: step.completedAt || null, + }; + } + + async updatePlanStep( + boardId: string, + initiativeId: string, + planId: string, + stepId: string, + data: Partial>, + ): Promise { + await this.ensureBoardStructure(boardId); + + const current = await this.getPlanStep(boardId, initiativeId, planId, stepId); + if (!current) return null; + + const now = timestamp(); + const nextStatus = data.status || current.status; + const completedAt = + nextStatus === "done" ? now : data.status && data.status !== "done" ? null : current.completedAt; + + const updated: PlanStep = { + ...current, + ...data, + status: nextStatus, + updatedAt: now, + completedAt, + assigneeAgentId: data.assigneeAgentId === undefined ? current.assigneeAgentId : data.assigneeAgentId, + order: data.order ?? current.order, + }; + + await this.store.put(planStepFile(boardId, initiativeId, planId, stepId), updated); + + const statusChange = data.status && data.status !== current.status; + await this.appendActivity(boardId, { + type: statusChange ? "plan_step.status_changed" : "plan_step.updated", + message: statusChange + ? `Plan step ${updated.title} moved ${current.status} -> ${updated.status}` + : `Plan step ${updated.title} updated`, + initiativeId, + payload: { planId, stepId, status: updated.status, fields: Object.keys(data) }, + agentId: updated.assigneeAgentId || undefined, + }); + + return updated; + } + + async deletePlanStep(boardId: string, initiativeId: string, planId: string, stepId: string): Promise { + await this.ensureBoardStructure(boardId); + + if (!(await this.store.exists(planStepFile(boardId, initiativeId, planId, stepId)))) return false; + + // Detach tasks FIRST, then delete the step — prevents orphaned references. + const tasks = await this.listTasks(boardId, initiativeId); + for (const task of tasks) { + if (task.planId === planId && task.planStepId === stepId) { + const detached: Task = { ...task, planStepId: null, updatedAt: timestamp() }; + await this.store.put(taskFile(boardId, initiativeId, task.id), detached); + } + } + + await this.store.delete(planStepFile(boardId, initiativeId, planId, stepId)); + + await this.appendActivity(boardId, { + type: "plan_step.removed", + message: `Plan step ${stepId} removed`, + initiativeId, + payload: { planId, stepId }, + }); + + return true; + } + + // --- Projects (legacy compatibility wrappers) --- + + async createProject(boardId: string, data: Pick): Promise { + return this.createInitiative(boardId, data); + } + + async listProjects(boardId: string): Promise { + return this.listInitiatives(boardId); + } + + async getProject(boardId: string, projectId: string): Promise { + return this.getInitiative(boardId, projectId); + } + + async updateProject( + boardId: string, + projectId: string, + data: Partial>, + ): Promise { + return this.updateInitiative(boardId, projectId, data); + } + + async deleteProject(boardId: string, projectId: string): Promise { + return this.deleteInitiative(boardId, projectId); + } + + // --- Tasks --- + + async createTask( + boardId: string, + initiativeId: string, + data: Pick & + Partial>, + actorAgentId: string, + ): Promise { + await this.ensureBoardStructure(boardId); + await this.requireRegisteredAgent(boardId, actorAgentId); + + const initiative = await this.getInitiative(boardId, initiativeId); + if (!initiative) { + throw new Error(`Initiative not found: ${initiativeId}`); + } + if (data.planId) { + const plan = await this.getPlan(boardId, initiativeId, data.planId); + if (!plan) { + throw new Error(`Plan not found: ${data.planId}`); + } + if (data.planStepId) { + const step = await this.getPlanStep(boardId, initiativeId, data.planId, data.planStepId); + if (!step) { + throw new Error(`Plan step not found: ${data.planStepId}`); + } + } + } else if (data.planStepId) { + throw new Error("planStepId requires a planId"); + } + + const baseId = slugify(data.title); + const id = await this.resolveUniqueId((candidate) => taskFile(boardId, initiativeId, candidate), baseId); + const now = timestamp(); + + const task = normalizeTask( + { + id, + initiativeId, + projectId: initiativeId, + planId: data.planId || null, + planStepId: data.planStepId || null, + boardId, + title: data.title, + description: data.description || "", + status: "todo", + assigneeAgentId: data.assigneeAgentId || null, + assigneeAgentIds: data.assigneeAgentIds || [], + deliverables: data.deliverables || [], + priority: data.priority || "medium", + tags: data.tags || [], + createdAt: now, + updatedAt: now, + completedAt: null, + }, + initiativeId, + ); + + await this.store.put(taskFile(boardId, initiativeId, id), task); + + await this.appendActivity(boardId, { + type: "task.created", + message: `Task ${task.title} created`, + initiativeId, + taskId: task.id, + agentId: actorAgentId, + payload: { + assignees: task.assigneeAgentIds, + status: task.status, + }, + }); + + return task; + } + + async listTasks( + boardId: string, + initiativeId: string, + filters?: { status?: string; assignee?: string; tag?: string }, + ): Promise { + await this.ensureBoardStructure(boardId); + + const files = await this.store.listChildren(tasksDir(boardId, initiativeId)); + let tasks: Task[] = []; + + for (const f of files) { + if (!f.endsWith(".json")) continue; + const task = await this.store.get(`${tasksDir(boardId, initiativeId)}/${f}`, isEntityRecord); + if (task) { + tasks.push(normalizeTask(task, initiativeId)); + } + } + + if (filters?.status) tasks = tasks.filter((t) => t.status === filters.status); + if (filters?.assignee) { + tasks = tasks.filter( + (t) => t.assigneeAgentId === filters.assignee || t.assigneeAgentIds.includes(filters.assignee as string), + ); + } + if (filters?.tag) tasks = tasks.filter((t) => t.tags.includes(filters.tag as string)); + + return tasks.sort((a, b) => compareStr(a.createdAt, b.createdAt)); + } + + async listAllBoardTasks( + boardId: string, + filters?: { status?: string; assignee?: string; tag?: string }, + ): Promise { + const initiatives = await this.listInitiatives(boardId); + const allTasks: Task[] = []; + + for (const initiative of initiatives) { + const tasks = await this.listTasks(boardId, initiative.id, filters); + allTasks.push(...tasks); + } + + return allTasks.sort((a, b) => compareStr(b.updatedAt, a.updatedAt)); + } + + async getTask(boardId: string, initiativeId: string, taskId: string): Promise { + await this.ensureBoardStructure(boardId); + + const task = await this.store.get(taskFile(boardId, initiativeId, taskId), isEntityRecord); + if (!task) return null; + + return normalizeTask(task, initiativeId); + } + + async updateTask( + boardId: string, + initiativeId: string, + taskId: string, + data: Partial< + Pick< + Task, + "title" | "description" | "status" | "planId" | "planStepId" | "assigneeAgentId" | "assigneeAgentIds" | "deliverables" | "priority" | "tags" + > + >, + actorAgentId: string, + ): Promise { + await this.ensureBoardStructure(boardId); + await this.requireRegisteredAgent(boardId, actorAgentId); + + const current = await this.getTask(boardId, initiativeId, taskId); + if (!current) return null; + + const nextPlanId = data.planId === undefined ? current.planId : data.planId; + const nextPlanStepId = data.planStepId === undefined ? current.planStepId : data.planStepId; + + if (nextPlanId) { + const plan = await this.getPlan(boardId, initiativeId, nextPlanId); + if (!plan) { + throw new Error(`Plan not found: ${nextPlanId}`); + } + if (nextPlanStepId) { + const step = await this.getPlanStep(boardId, initiativeId, nextPlanId, nextPlanStepId); + if (!step) { + throw new Error(`Plan step not found: ${nextPlanStepId}`); + } + } + } else if (nextPlanStepId) { + throw new Error("planStepId requires a planId"); + } + + const now = timestamp(); + const nextStatus = data.status || current.status; + const completedAt = + nextStatus === "done" ? now : data.status && data.status !== "done" ? null : current.completedAt; + + const updated = normalizeTask( + { + ...current, + ...data, + status: nextStatus, + updatedAt: now, + completedAt, + }, + initiativeId, + ); + + await this.store.put(taskFile(boardId, initiativeId, taskId), updated); + await this.syncAgentPointersForTask(boardId, updated); + + const changedFields = Object.keys(data); + const statusChange = data.status && data.status !== current.status; + + await this.appendActivity(boardId, { + type: statusChange ? "task.status_changed" : "task.updated", + message: statusChange + ? `Task ${updated.title} moved ${current.status} -> ${updated.status}` + : `Task ${updated.title} updated`, + initiativeId, + taskId, + agentId: actorAgentId, + payload: { fields: changedFields, status: updated.status }, + }); + + return updated; + } + + async deleteTask(boardId: string, initiativeId: string, taskId: string, actorAgentId: string): Promise { + await this.ensureBoardStructure(boardId); + await this.requireRegisteredAgent(boardId, actorAgentId); + + // Read first: getTask may quarantine a corrupt file (moving it aside), so + // capture the task before any delete and tolerate an already-gone file. + const task = await this.getTask(boardId, initiativeId, taskId); + if (!task && !(await this.store.exists(taskFile(boardId, initiativeId, taskId)))) return false; + + await this.store.delete(taskFile(boardId, initiativeId, taskId)); + + if (task) { + await this.syncAgentPointersForTask(boardId, { ...task, status: "todo" as TaskStatus }); + await this.appendActivity(boardId, { + type: "task.removed", + message: `Task ${task.title} removed`, + initiativeId, + taskId, + agentId: actorAgentId, + }); + } + + return true; + } + + // --- Activity --- + + async listActivity( + boardId: string, + options?: { limit?: number; initiativeId?: string; agentId?: string; taskId?: string }, + ): Promise { + await this.ensureBoardStructure(boardId); + + const buckets = (await this.store.listEventBuckets(eventsDir(boardId))).sort((a, b) => b.localeCompare(a)); + + const limit = options?.limit ?? 200; + const hasFilters = !!(options?.initiativeId || options?.agentId || options?.taskId); + const events: ActivityEvent[] = []; + + const matchesFilters = (event: ActivityEvent): boolean => { + if (options?.initiativeId && event.initiativeId !== options.initiativeId) return false; + if (options?.agentId && event.agentId !== options.agentId) return false; + if (options?.taskId && event.taskId !== options.taskId) return false; + return true; + }; + + for (let i = 0; i < buckets.length; i++) { + // Buckets are sorted descending by date — stop early when no filters active. + if (!hasFilters && events.length >= limit) break; + + const records = await this.store.readEventBucket(eventsDir(boardId), buckets[i]!); + for (const record of records) { + if (!isRecord(record)) continue; + const event = record as unknown as ActivityEvent; + if (hasFilters && !matchesFilters(event)) continue; + events.push(event); + } + + // With filters, stop once we have enough matches and older buckets cannot + // contribute events newer than the current cutoff. + if (hasFilters && events.length >= limit) { + events.sort((a, b) => compareStr(b.createdAt, a.createdAt)); + const cutoffDate = events[limit - 1]!.createdAt.slice(0, 10); + const nextBucket = buckets[i + 1]; + if (!nextBucket || nextBucket < cutoffDate) break; + } + } + + events.sort((a, b) => compareStr(b.createdAt, a.createdAt)); + + return events.slice(0, limit); + } +} diff --git a/src/lib/storage/fs-object-store.ts b/src/lib/storage/fs-object-store.ts new file mode 100644 index 0000000..bc271b1 --- /dev/null +++ b/src/lib/storage/fs-object-store.ts @@ -0,0 +1,170 @@ +import { appendFile, mkdir, readdir, readFile, rename, rm, writeFile } from "fs/promises"; +import { existsSync } from "fs"; +import path from "path"; +import { randomUUID } from "crypto"; +import { getDataDir } from "@/lib/server-utils"; +import { timestamp } from "@/lib/utils"; +import type { ObjectStore } from "./object-store"; + +function hasErrnoCode(err: unknown, code: string): boolean { + return typeof err === "object" && err !== null && "code" in err && (err as { code: string }).code === code; +} + +// --- Per-file write serialization --- +// Concurrent writers to the same path must not interleave their write+rename +// sequences. We chain operations per absolute path so each observes a +// consistent on-disk state; stale entries are pruned once their tail settles. +const fileWriteQueues = new Map>(); + +function withFileLock(key: string, fn: () => Promise): Promise { + const previous = fileWriteQueues.get(key) ?? Promise.resolve(); + const run = previous.then(fn, fn); + const settled = run.then( + () => undefined, + () => undefined, + ); + fileWriteQueues.set(key, settled); + void settled.then(() => { + if (fileWriteQueues.get(key) === settled) { + fileWriteQueues.delete(key); + } + }); + return run; +} + +function quarantineDir(): string { + return path.join(getDataDir(), ".quarantine"); +} + +async function quarantineFile(filePath: string, reason: string): Promise { + try { + const dir = quarantineDir(); + await mkdir(dir, { recursive: true }); + const stamp = timestamp().replace(/[:.]/g, "-"); + const dest = path.join(dir, `${stamp}-${randomUUID().slice(0, 8)}-${path.basename(filePath)}`); + await rename(filePath, dest); + console.error(`Quarantined corrupt file ${filePath} -> ${dest} (${reason})`); + } catch (err) { + console.error(`Failed to quarantine ${filePath}:`, err); + } +} + +/** + * Filesystem-backed ObjectStore. Keys are POSIX-relative paths mapped under the + * configured data dir. Suitable for local / self-hosted use; provides atomic + * writes, per-path serialization, and corrupt-file quarantine. + */ +export class FsObjectStore implements ObjectStore { + private abs(key: string): string { + return path.join(getDataDir(), key); + } + + async get(key: string, validate?: (value: unknown) => boolean): Promise { + const filePath = this.abs(key); + let content: string; + try { + content = await readFile(filePath, "utf-8"); + } catch (err: unknown) { + if (hasErrnoCode(err, "ENOENT")) return null; + throw err; + } + + let parsed: unknown; + try { + parsed = JSON.parse(content); + } catch (err) { + console.error(`Corrupted JSON in ${filePath}:`, (err as Error).message); + await quarantineFile(filePath, "invalid JSON"); + return null; + } + + if (validate && !validate(parsed)) { + console.error(`Schema validation failed for ${filePath}; skipping`); + return null; + } + + return parsed as T; + } + + async put(key: string, data: unknown): Promise { + const filePath = this.abs(key); + await withFileLock(filePath, async () => { + await mkdir(path.dirname(filePath), { recursive: true }); + // Unique temp path per write so concurrent writers never share a stager. + const tmpPath = `${filePath}.${process.pid}.${randomUUID()}.tmp`; + try { + await writeFile(tmpPath, JSON.stringify(data, null, 2), "utf-8"); + await rename(tmpPath, filePath); + } catch (err) { + await rm(tmpPath, { force: true }).catch(() => {}); + throw err; + } + }); + } + + async createIfAbsent(key: string): Promise { + const filePath = this.abs(key); + await mkdir(path.dirname(filePath), { recursive: true }); + try { + // O_CREAT|O_EXCL: fail if the file already exists — eliminates TOCTOU race. + await writeFile(filePath, "{}", { flag: "wx" }); + return true; + } catch (err: unknown) { + if (hasErrnoCode(err, "EEXIST")) return false; + throw err; + } + } + + async exists(key: string): Promise { + return existsSync(this.abs(key)); + } + + async delete(key: string): Promise { + await rm(this.abs(key), { force: true }); + } + + async deletePrefix(prefix: string): Promise { + await rm(this.abs(prefix), { recursive: true, force: true }); + } + + async listChildren(prefix: string): Promise { + try { + return await readdir(this.abs(prefix)); + } catch { + return []; + } + } + + private eventFile(eventsKey: string, bucket: string): string { + return this.abs(`${eventsKey}/${bucket}.ndjson`); + } + + async appendEvent(eventsKey: string, bucket: string, record: unknown): Promise { + const filePath = this.eventFile(eventsKey, bucket); + // Serialize appends to the same daily file so concurrent events cannot + // interleave partial lines. + await withFileLock(filePath, async () => { + await mkdir(path.dirname(filePath), { recursive: true }); + await appendFile(filePath, `${JSON.stringify(record)}\n`, "utf-8"); + }); + } + + async listEventBuckets(eventsKey: string): Promise { + const files = await this.listChildren(eventsKey); + return files.filter((name) => name.endsWith(".ndjson")).map((name) => name.slice(0, -".ndjson".length)); + } + + async readEventBucket(eventsKey: string, bucket: string): Promise { + const content = await readFile(this.eventFile(eventsKey, bucket), "utf-8").catch(() => ""); + if (!content) return []; + const records: unknown[] = []; + for (const line of content.split("\n").filter(Boolean)) { + try { + records.push(JSON.parse(line)); + } catch { + // Skip malformed lines and continue. + } + } + return records; + } +} diff --git a/src/lib/storage/fs-storage.ts b/src/lib/storage/fs-storage.ts index 0031240..1afbf31 100644 --- a/src/lib/storage/fs-storage.ts +++ b/src/lib/storage/fs-storage.ts @@ -1,1554 +1,60 @@ -import { appendFile, mkdir, readdir, readFile, rename, rm, writeFile } from "fs/promises"; -import { existsSync } from "fs"; -import path from "path"; -import { randomUUID } from "crypto"; -import type { - ActivityEvent, - Agent, - Board, - BoardSummary, - Initiative, - Plan, - PlanStep, - Project, - Task, - TaskStatus, -} from "@/lib/types"; import type { Storage } from "./index"; -import { slugify, timestamp } from "@/lib/utils"; -import { getDataDir } from "@/lib/server-utils"; +import type { ObjectStore } from "./object-store"; +import { DocumentStorage, AgentRegistrationError } from "./document-storage"; +import { FsObjectStore } from "./fs-object-store"; -type AgentIntro = { - runtime: string; - sessionKey: string; - model: string; - thread: { - id: string; - name: string; - source?: string; - }; - workingDirectory: string; - host?: { - hostname: string; - localIp: string; - }; -}; - -export class AgentRegistrationError extends Error { - code: string; - - constructor(code: string, message: string) { - super(message); - this.name = "AgentRegistrationError"; - this.code = code; - } -} - -function isRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null && !Array.isArray(value); -} - -function normalizeAgentName(name: string): string { - return name.trim(); -} - -function canonicalAgentName(name: string): string { - return normalizeAgentName(name).toLowerCase(); -} - -function normalizeText(value: unknown): string { - return typeof value === "string" ? value.trim() : ""; -} - -function inferModelFromRuntime(runtime: string): string { - const key = runtime.toLowerCase(); - if (key.includes("codex") || key.includes("openai")) return "codex"; - if (key.includes("claude") || key.includes("anthropic")) return "claude"; - if (key.includes("cursor")) return "cursor"; - return "unknown-model"; -} - -function extractThreadContext(intro: Record): AgentIntro["thread"] | null { - const threadRaw = intro.thread; - - if (typeof threadRaw === "string") { - const id = normalizeText(threadRaw); - const name = normalizeText(intro.threadName) || id; - if (!id) return null; - return { id, name }; - } - - if (!isRecord(threadRaw)) return null; - - const id = normalizeText(threadRaw.id) || normalizeText(intro.threadId); - const name = normalizeText(threadRaw.name) || normalizeText(intro.threadName) || id; - const source = normalizeText(threadRaw.source); - if (!id) return null; - return source ? { id, name, source } : { id, name }; -} - -function extractAgentIntro(metadata: unknown): AgentIntro | null { - if (!isRecord(metadata)) return null; - const intro = metadata.intro; - if (!isRecord(intro)) return null; - - const runtime = normalizeText(intro.runtime) || "unknown-runtime"; - const sessionKey = normalizeText(intro.sessionKey) || normalizeText(intro.instanceKey); - const model = normalizeText(intro.model) || inferModelFromRuntime(runtime); - const thread = - extractThreadContext(intro) || - (() => { - const fallbackId = normalizeText(intro.threadId) || "unknown-thread"; - const fallbackName = normalizeText(intro.threadName) || fallbackId; - return { id: fallbackId, name: fallbackName, source: runtime }; - })(); - const workingDirectory = normalizeText(intro.workingDirectory) || "unknown-working-directory"; - const hostRaw = isRecord(intro.host) ? intro.host : {}; - const hostHostname = normalizeText(hostRaw.hostname); - const hostLocalIp = normalizeText(hostRaw.localIp); - const host = - hostHostname || hostLocalIp - ? { - hostname: hostHostname || "unknown-host", - localIp: hostLocalIp || "unknown-local-ip", - } - : undefined; +export { AgentRegistrationError }; - if (!sessionKey) return null; - return { runtime, sessionKey, model, thread, workingDirectory, host }; -} - -function hasErrnoCode(err: unknown, code: string): boolean { - return typeof err === "object" && err !== null && "code" in err && (err as { code: string }).code === code; -} - -// --- Per-file write serialization --- -// Concurrent writers to the same logical path must not interleave their -// write+rename sequences. We chain operations per absolute path so each one -// observes a consistent on-disk state. The chain continues regardless of -// whether a prior operation resolved or rejected, and stale entries are pruned -// once their tail settles to avoid unbounded map growth. -const fileWriteQueues = new Map>(); - -function withFileLock(key: string, fn: () => Promise): Promise { - const previous = fileWriteQueues.get(key) ?? Promise.resolve(); - const run = previous.then(fn, fn); - const settled = run.then( - () => undefined, - () => undefined, - ); - fileWriteQueues.set(key, settled); - void settled.then(() => { - if (fileWriteQueues.get(key) === settled) { - fileWriteQueues.delete(key); - } - }); - return run; -} - -// --- Corrupt-file quarantine --- -// A file whose bytes cannot be parsed as JSON is unrecoverable for the running -// process. Rather than repeatedly failing to read it (or, worse, letting a -// caller crash on it), we move it aside into a quarantine directory so the -// system keeps running and a human can inspect the bad data later. -function quarantineDir(): string { - return path.join(getDataDir(), ".quarantine"); -} - -async function quarantineFile(filePath: string, reason: string): Promise { - try { - const dir = quarantineDir(); - await mkdir(dir, { recursive: true }); - const stamp = timestamp().replace(/[:.]/g, "-"); - const dest = path.join(dir, `${stamp}-${randomUUID().slice(0, 8)}-${path.basename(filePath)}`); - await rename(filePath, dest); - console.error(`Quarantined corrupt file ${filePath} -> ${dest} (${reason})`); - } catch (err) { - // Quarantine is best-effort; never let it mask the original read outcome. - console.error(`Failed to quarantine ${filePath}:`, err); +/** + * Filesystem-backed storage (local / self-hosted). Retained as a named export + * for direct construction (e.g. tests) and backward compatibility. + */ +export class FsStorage extends DocumentStorage { + constructor() { + super(new FsObjectStore()); } } -// A stored entity is only usable if it parsed to an object carrying a string id. -// Anything else is either a transient reservation placeholder (see -// resolveUniqueId) or structurally unusable, and is skipped on read. -function isEntityRecord(value: unknown): boolean { - return isRecord(value) && typeof value.id === "string" && value.id.length > 0; -} - /** - * Reads and parses a JSON file. - * - * - Missing file (ENOENT) -> null. - * - Unparseable bytes -> the file is quarantined and null is returned, so a - * single corrupt file can never crash or wedge a read path. - * - When a `validate` guard is supplied, a parsed value that fails validation - * is skipped (null) but left in place — it may be a newer/older schema or an - * in-flight reservation placeholder that we should not destroy. + * Picks the storage backend: + * - `AGENTBOARD_STORAGE=blob|fs` forces a backend explicitly. + * - Otherwise, a present `BLOB_READ_WRITE_TOKEN` (the Vercel Blob env var) + * selects Blob, since the filesystem is read-only on serverless. + * - Falls back to the filesystem for local development. */ -async function readJson(filePath: string, validate?: (value: unknown) => boolean): Promise { - let content: string; - try { - content = await readFile(filePath, "utf-8"); - } catch (err: unknown) { - if (hasErrnoCode(err, "ENOENT")) return null; - throw err; - } - - let parsed: unknown; - try { - parsed = JSON.parse(content); - } catch (err) { - console.error(`Corrupted JSON in ${filePath}:`, (err as Error).message); - await quarantineFile(filePath, "invalid JSON"); - return null; +async function selectObjectStore(): Promise { + const mode = process.env.AGENTBOARD_STORAGE?.toLowerCase(); + if (mode === "blob") { + const { BlobObjectStore } = await import("./blob-object-store"); + return new BlobObjectStore(); } - - if (validate && !validate(parsed)) { - console.error(`Schema validation failed for ${filePath}; skipping`); - return null; - } - - return parsed as T; -} - -async function writeJson(filePath: string, data: unknown): Promise { - // Serialize per final path so concurrent writers cannot clobber each other. - await withFileLock(filePath, async () => { - await mkdir(path.dirname(filePath), { recursive: true }); - // Unique temp path per write: a shared `${filePath}.tmp` would be raced by - // concurrent writers, producing a torn/garbage final file after rename. - const tmpPath = `${filePath}.${process.pid}.${randomUUID()}.tmp`; - try { - await writeFile(tmpPath, JSON.stringify(data, null, 2), "utf-8"); - await rename(tmpPath, filePath); - } catch (err) { - await rm(tmpPath, { force: true }).catch(() => {}); - throw err; - } - }); -} - -async function listDir(dirPath: string): Promise { - try { - return await readdir(dirPath); - } catch { - return []; + if (mode === "fs") return new FsObjectStore(); + if (process.env.BLOB_READ_WRITE_TOKEN) { + const { BlobObjectStore } = await import("./blob-object-store"); + return new BlobObjectStore(); } + return new FsObjectStore(); } -function boardDir(boardId: string): string { - return path.join(getDataDir(), "boards", boardId); -} - -function agentsDir(boardId: string): string { - return path.join(boardDir(boardId), "agents"); -} - -function initiativesDir(boardId: string): string { - return path.join(boardDir(boardId), "initiatives"); -} - -function initiativeFile(boardId: string, initiativeId: string): string { - return path.join(initiativesDir(boardId), `${initiativeId}.json`); -} - -function tasksDir(boardId: string, initiativeId: string): string { - return path.join(initiativesDir(boardId), initiativeId, "tasks"); -} - -function taskFile(boardId: string, initiativeId: string, taskId: string): string { - return path.join(tasksDir(boardId, initiativeId), `${taskId}.json`); -} - -function plansDir(boardId: string, initiativeId: string): string { - return path.join(initiativesDir(boardId), initiativeId, "plans"); -} - -function planFile(boardId: string, initiativeId: string, planId: string): string { - return path.join(plansDir(boardId, initiativeId), `${planId}.json`); -} - -function planStepsDir(boardId: string, initiativeId: string, planId: string): string { - return path.join(plansDir(boardId, initiativeId), planId, "steps"); -} - -function planStepFile(boardId: string, initiativeId: string, planId: string, stepId: string): string { - return path.join(planStepsDir(boardId, initiativeId, planId), `${stepId}.json`); -} - -function eventsDir(boardId: string): string { - return path.join(boardDir(boardId), "events"); -} - -function legacyProjectsDir(boardId: string): string { - return path.join(boardDir(boardId), "projects"); -} - -function migrationMarkerFile(boardId: string): string { - return path.join(boardDir(boardId), ".migration-v2.json"); -} +// Singleton — use globalThis to survive Next.js HMR in development. +const globalForStorage = globalThis as unknown as { + _agentboardStorage?: Storage; + _agentboardStoragePromise?: Promise; +}; -async function resolveUniqueId(filePathForId: (id: string) => string, desiredId: string): Promise { - let candidate = desiredId; - let counter = 2; - for (;;) { - const filePath = filePathForId(candidate); - await mkdir(path.dirname(filePath), { recursive: true }); - try { - // O_CREAT|O_EXCL: fail if file already exists — eliminates TOCTOU race - await writeFile(filePath, "{}", { flag: "wx" }); - return candidate; - } catch (err: unknown) { - if (typeof err === "object" && err !== null && "code" in err && (err as { code: string }).code === "EEXIST") { - candidate = `${desiredId}-${counter}`; - counter += 1; - continue; - } +export async function getStorage(): Promise { + if (globalForStorage._agentboardStorage) { + return globalForStorage._agentboardStorage; + } + if (!globalForStorage._agentboardStoragePromise) { + globalForStorage._agentboardStoragePromise = (async () => { + const storage = new DocumentStorage(await selectObjectStore()); + globalForStorage._agentboardStorage = storage; + return storage; + })().catch((err) => { + globalForStorage._agentboardStoragePromise = undefined; throw err; - } - } -} - -// Defensive string compare for sort keys that may be missing on partially -// written or legacy records — `undefined.localeCompare` would throw and take -// down an entire list endpoint. -function compareStr(a: string | undefined, b: string | undefined): number { - return (a ?? "").localeCompare(b ?? ""); -} - -function normalizeAssignees(task: Partial): string[] { - if (task.assigneeAgentIds && task.assigneeAgentIds.length > 0) { - return Array.from(new Set(task.assigneeAgentIds.filter(Boolean))); - } - if (task.assigneeAgentId) { - return [task.assigneeAgentId]; - } - return []; -} - -function normalizeTask(task: Task, fallbackInitiativeId?: string): Task { - const initiativeId = task.initiativeId || task.projectId || fallbackInitiativeId || "general"; - const assigneeAgentIds = normalizeAssignees(task); - const assigneeAgentId = task.assigneeAgentId || assigneeAgentIds[0] || null; - - return { - ...task, - initiativeId, - projectId: task.projectId || initiativeId, - planId: task.planId || null, - planStepId: task.planStepId || null, - assigneeAgentId, - assigneeAgentIds, - deliverables: task.deliverables || [], - }; -} - -export class FsStorage implements Storage { - private initializedBoards = new Set(); - private initializingBoards = new Map>(); - - private async ensureBoardStructure(boardId: string): Promise { - if (this.initializedBoards.has(boardId)) return; - const existingInit = this.initializingBoards.get(boardId); - if (existingInit) { - await existingInit; - return; - } - - const initPromise = (async () => { - await mkdir(agentsDir(boardId), { recursive: true }); - await mkdir(initiativesDir(boardId), { recursive: true }); - await mkdir(eventsDir(boardId), { recursive: true }); - - await this.migrateLegacyProjects(boardId); - this.initializedBoards.add(boardId); - })(); - - this.initializingBoards.set(boardId, initPromise); - try { - await initPromise; - } finally { - this.initializingBoards.delete(boardId); - } - } - - private async migrateLegacyProjects(boardId: string): Promise { - if (existsSync(migrationMarkerFile(boardId))) return; - - const projectIds = await listDir(legacyProjectsDir(boardId)); - if (projectIds.length === 0) { - await writeJson(migrationMarkerFile(boardId), { version: 2, migratedAt: timestamp(), migratedProjects: 0 }); - return; - } - - let migratedProjects = 0; - let migratedTasks = 0; - - for (const projectId of projectIds) { - const legacyProjectPath = path.join(legacyProjectsDir(boardId), projectId, "project.json"); - const legacyProject = await readJson(legacyProjectPath, isEntityRecord); - if (!legacyProject) continue; - - const nextInitiative: Initiative = { - id: legacyProject.id, - boardId, - name: legacyProject.name, - description: legacyProject.description || "", - status: "active", - kind: "general", - assigneeAgentIds: [], - createdAt: legacyProject.createdAt || timestamp(), - updatedAt: legacyProject.updatedAt || legacyProject.createdAt || timestamp(), - }; - - if (!existsSync(initiativeFile(boardId, projectId))) { - await writeJson(initiativeFile(boardId, projectId), nextInitiative); - } - - const legacyTaskDir = path.join(legacyProjectsDir(boardId), projectId, "tasks"); - const taskFiles = await listDir(legacyTaskDir); - - for (const taskName of taskFiles) { - if (!taskName.endsWith(".json")) continue; - const legacyTask = await readJson(path.join(legacyTaskDir, taskName), isEntityRecord); - if (!legacyTask) continue; - - const normalized = normalizeTask(legacyTask, projectId); - const nextTaskPath = taskFile(boardId, projectId, normalized.id); - if (!existsSync(nextTaskPath)) { - await writeJson(nextTaskPath, normalized); - migratedTasks += 1; - } - } - - migratedProjects += 1; - } - - await writeJson(migrationMarkerFile(boardId), { - version: 2, - migratedAt: timestamp(), - migratedProjects, - migratedTasks, - }); - - await this.appendActivity( - boardId, - { - type: "system.migration.v2", - message: `Migrated ${migratedProjects} legacy project(s) and ${migratedTasks} task(s) to initiatives`, - payload: { migratedProjects, migratedTasks }, - }, - { skipEnsure: true }, - ); - } - - private async appendActivity( - boardId: string, - event: Omit, - options?: { skipEnsure?: boolean }, - ): Promise { - if (!options?.skipEnsure) { - await this.ensureBoardStructure(boardId); - } - - const createdAt = timestamp(); - const activity: ActivityEvent = { - id: randomUUID(), - boardId, - createdAt, - ...event, - }; - - const dailyFile = path.join(eventsDir(boardId), `${createdAt.slice(0, 10)}.ndjson`); - // Serialize appends to the same daily file so concurrent events cannot - // interleave partial lines; listActivity already tolerates the rare torn - // line, but serialization keeps the common case clean. - await withFileLock(dailyFile, async () => { - await mkdir(eventsDir(boardId), { recursive: true }); - await appendFile(dailyFile, `${JSON.stringify(activity)}\n`, "utf-8"); - }); - - return activity; - } - - private async writeAgent(boardId: string, agent: Agent): Promise { - await writeJson(path.join(agentsDir(boardId), `${agent.id}.json`), agent); - } - - private async requireRegisteredAgent(boardId: string, agentId: string): Promise { - const agent = await this.getAgent(boardId, agentId); - if (!agent) { - throw new Error(`Agent not registered: ${agentId}`); - } - return agent; - } - - private async syncAgentPointersForTask(boardId: string, task: Task): Promise { - const assignees = normalizeAssignees(task); - - if (task.status === "in_progress") { - for (const agentId of assignees) { - const agent = await this.getAgent(boardId, agentId); - if (!agent) continue; - const updated: Agent = { - ...agent, - currentTaskId: task.id, - currentInitiativeId: task.initiativeId, - lastHeartbeat: timestamp(), - }; - await this.writeAgent(boardId, updated); - } - return; - } - - const agents = await this.listAgents(boardId); - for (const agent of agents) { - if (agent.currentTaskId !== task.id || agent.currentInitiativeId !== task.initiativeId) { - continue; - } - const cleared: Agent = { - ...agent, - currentTaskId: null, - currentInitiativeId: null, - lastHeartbeat: timestamp(), - }; - await this.writeAgent(boardId, cleared); - } - } - - // --- Boards --- - - async createBoard(data: Pick): Promise { - const id = slugify(data.name); - const now = timestamp(); - const board: Board = { - id, - name: data.name, - description: data.description || "", - createdAt: now, - updatedAt: now, - }; - - await mkdir(agentsDir(id), { recursive: true }); - await mkdir(initiativesDir(id), { recursive: true }); - await mkdir(eventsDir(id), { recursive: true }); - await writeJson(path.join(boardDir(id), "board.json"), board); - - return board; - } - - async listBoards(): Promise { - const boardsDir = path.join(getDataDir(), "boards"); - const dirs = await listDir(boardsDir); - const boards: Board[] = []; - - for (const d of dirs) { - const board = await readJson(path.join(boardsDir, d, "board.json"), isEntityRecord); - if (board) boards.push(board); - } - - return boards.sort((a, b) => compareStr(b.createdAt, a.createdAt)); - } - - async getBoard(boardId: string): Promise { - return readJson(path.join(boardDir(boardId), "board.json"), isEntityRecord); - } - - async getBoardSummary(boardId: string): Promise { - await this.ensureBoardStructure(boardId); - - const board = await this.getBoard(boardId); - if (!board) return null; - - // Isolate each section: a failure (corrupt subtree, transient FS error) - // in one count must not blank out the whole summary. Each section falls - // back to an empty result and the board still renders. - const safeSection = async (label: string, fn: () => Promise, fallback: T): Promise => { - try { - return await fn(); - } catch (err) { - console.error(`Board summary section "${label}" failed for ${boardId}:`, err); - return fallback; - } - }; - - const [agents, initiatives, tasks] = await Promise.all([ - safeSection("agents", () => this.listAgents(boardId), [] as Agent[]), - safeSection("initiatives", () => this.listInitiatives(boardId), [] as Initiative[]), - safeSection("tasks", () => this.listAllBoardTasks(boardId), [] as Task[]), - ]); - - // Parallelize plan reads across initiatives - const plansByInitiative = await Promise.all( - initiatives.map((initiative) => - safeSection(`plans:${initiative.id}`, () => this.listPlans(boardId, initiative.id), [] as Plan[]), - ), - ); - - const allPlans = plansByInitiative.flat(); - - // Parallelize step counts across plans - const stepsByPlan = await Promise.all( - allPlans.map((plan) => - safeSection( - `steps:${plan.id}`, - () => this.listPlanSteps(boardId, plan.initiativeId, plan.id), - [] as PlanStep[], - ), - ), - ); - - const planStepCount = stepsByPlan.reduce((sum, steps) => sum + steps.length, 0); - - return { - ...board, - agentCount: agents.length, - initiativeCount: initiatives.length, - planCount: allPlans.length, - planStepCount, - projectCount: initiatives.length, - taskCount: tasks.length, - }; - } - - async updateBoard(boardId: string, data: Partial>): Promise { - await this.ensureBoardStructure(boardId); - - const board = await this.getBoard(boardId); - if (!board) return null; - - const updated = { ...board, ...data, updatedAt: timestamp() }; - await writeJson(path.join(boardDir(boardId), "board.json"), updated); - - await this.appendActivity(boardId, { - type: "board.updated", - message: "Board details updated", - payload: { fields: Object.keys(data) }, - }); - - return updated; - } - - async deleteBoard(boardId: string): Promise { - const dir = boardDir(boardId); - if (!existsSync(dir)) return false; - await rm(dir, { recursive: true }); - this.initializedBoards.delete(boardId); - return true; - } - - // --- Agents --- - - async createAgent(boardId: string, data: Pick): Promise { - await this.ensureBoardStructure(boardId); - - const name = normalizeAgentName(data.name || ""); - if (!name) { - throw new AgentRegistrationError("MISSING_NAME", "Agent name is required"); - } - - const id = slugify(name); - if (!id) { - throw new AgentRegistrationError("INVALID_NAME", "Agent name must contain at least one letter or number"); - } - - const metadata = isRecord(data.metadata) ? data.metadata : {}; - const intro = extractAgentIntro(metadata); - if (!intro) { - throw new AgentRegistrationError( - "INVALID_AGENT_INTRO", - "Agent intro metadata requires sessionKey (runtime/model/thread/workingDirectory are auto-detected or defaulted)", - ); - } - - const existingById = await this.getAgent(boardId, id); - if (existingById) { - throw new AgentRegistrationError("AGENT_ID_CONFLICT", `Agent id already exists: ${id}`); - } - - const existingAgents = await this.listAgents(boardId); - const desiredName = canonicalAgentName(name); - for (const existing of existingAgents) { - if (canonicalAgentName(existing.name) === desiredName) { - throw new AgentRegistrationError("AGENT_NAME_CONFLICT", `Agent name already exists: ${name}`); - } - const existingIntro = extractAgentIntro(existing.metadata); - if (existingIntro?.sessionKey === intro.sessionKey) { - throw new AgentRegistrationError("AGENT_KEY_CONFLICT", `Agent key already exists: ${intro.sessionKey}`); - } - } - - const now = timestamp(); - - const agent: Agent = { - id, - boardId, - name, - description: data.description || "", - status: "idle", - statusMessage: "", - currentTaskId: null, - currentInitiativeId: null, - metadata, - registeredAt: now, - lastHeartbeat: now, - }; - - await this.writeAgent(boardId, agent); - - await this.appendActivity(boardId, { - type: "agent.registered", - message: `Agent ${agent.name} registered`, - agentId: agent.id, - }); - - return agent; - } - - async listAgents(boardId: string): Promise { - await this.ensureBoardStructure(boardId); - - const files = await listDir(agentsDir(boardId)); - const agents: Agent[] = []; - - for (const f of files) { - if (!f.endsWith(".json")) continue; - const agent = await readJson(path.join(agentsDir(boardId), f), isEntityRecord); - if (!agent) continue; - - agents.push({ - ...agent, - currentInitiativeId: agent.currentInitiativeId || null, - }); - } - - return agents.sort((a, b) => compareStr(a.name, b.name)); - } - - async getAgent(boardId: string, agentId: string): Promise { - await this.ensureBoardStructure(boardId); - - const agent = await readJson(path.join(agentsDir(boardId), `${agentId}.json`), isEntityRecord); - if (!agent) return null; - - return { - ...agent, - currentInitiativeId: agent.currentInitiativeId || null, - }; - } - - async updateAgent( - boardId: string, - agentId: string, - data: Partial>, - ): Promise { - await this.ensureBoardStructure(boardId); - - const agent = await this.getAgent(boardId, agentId); - if (!agent) return null; - - const updated: Agent = { - ...agent, - ...data, - lastHeartbeat: timestamp(), - }; - - await this.writeAgent(boardId, updated); - - await this.appendActivity(boardId, { - type: "agent.updated", - message: `Agent ${updated.name} updated status to ${updated.status}`, - agentId, - payload: { fields: Object.keys(data) }, - }); - - return updated; - } - - async deleteAgent(boardId: string, agentId: string): Promise { - await this.ensureBoardStructure(boardId); - - const filePath = path.join(agentsDir(boardId), `${agentId}.json`); - if (!existsSync(filePath)) return false; - await rm(filePath); - - await this.appendActivity(boardId, { - type: "agent.removed", - message: `Agent ${agentId} removed`, - agentId, }); - - return true; - } - - async heartbeatAgent(boardId: string, agentId: string, message?: string): Promise { - await this.ensureBoardStructure(boardId); - - const agent = await this.getAgent(boardId, agentId); - if (!agent) return null; - - const updated: Agent = { - ...agent, - status: "active", - lastHeartbeat: timestamp(), - ...(message !== undefined ? { statusMessage: message } : {}), - }; - - await this.writeAgent(boardId, updated); - - await this.appendActivity(boardId, { - type: "agent.heartbeat", - message: message ? `Heartbeat: ${message}` : "Heartbeat", - agentId, - initiativeId: updated.currentInitiativeId || undefined, - taskId: updated.currentTaskId || undefined, - }); - - return updated; - } - - // --- Initiatives --- - - async createInitiative( - boardId: string, - data: Pick & Partial>, - ): Promise { - await this.ensureBoardStructure(boardId); - - const baseId = slugify(data.name); - const id = await resolveUniqueId((candidate) => initiativeFile(boardId, candidate), baseId); - const now = timestamp(); - - const initiative: Initiative = { - id, - boardId, - name: data.name, - description: data.description || "", - status: data.status || "active", - kind: data.kind || "thread", - assigneeAgentIds: data.assigneeAgentIds || [], - createdAt: now, - updatedAt: now, - }; - - await writeJson(initiativeFile(boardId, id), initiative); - await mkdir(tasksDir(boardId, id), { recursive: true }); - await mkdir(plansDir(boardId, id), { recursive: true }); - - await this.appendActivity(boardId, { - type: "initiative.created", - message: `Initiative ${initiative.name} created`, - initiativeId: initiative.id, - payload: { kind: initiative.kind }, - }); - - return initiative; - } - - async listInitiatives(boardId: string): Promise { - await this.ensureBoardStructure(boardId); - - const files = await listDir(initiativesDir(boardId)); - const initiatives: Initiative[] = []; - - for (const f of files) { - if (!f.endsWith(".json")) continue; - const initiative = await readJson(path.join(initiativesDir(boardId), f), isEntityRecord); - if (initiative) { - initiatives.push({ - ...initiative, - status: initiative.status || "active", - kind: initiative.kind || "thread", - assigneeAgentIds: initiative.assigneeAgentIds || [], - }); - } - } - - return initiatives.sort((a, b) => compareStr(a.createdAt, b.createdAt)); - } - - async getInitiative(boardId: string, initiativeId: string): Promise { - await this.ensureBoardStructure(boardId); - - const initiative = await readJson(initiativeFile(boardId, initiativeId), isEntityRecord); - if (!initiative) return null; - - return { - ...initiative, - status: initiative.status || "active", - kind: initiative.kind || "thread", - assigneeAgentIds: initiative.assigneeAgentIds || [], - }; - } - - async updateInitiative( - boardId: string, - initiativeId: string, - data: Partial>, - ): Promise { - await this.ensureBoardStructure(boardId); - - const initiative = await this.getInitiative(boardId, initiativeId); - if (!initiative) return null; - - const updated: Initiative = { - ...initiative, - ...data, - updatedAt: timestamp(), - }; - - await writeJson(initiativeFile(boardId, initiativeId), updated); - - await this.appendActivity(boardId, { - type: "initiative.updated", - message: `Initiative ${updated.name} updated`, - initiativeId, - payload: { fields: Object.keys(data) }, - }); - - return updated; - } - - async deleteInitiative(boardId: string, initiativeId: string): Promise { - await this.ensureBoardStructure(boardId); - - const dir = path.join(initiativesDir(boardId), initiativeId); - const dataFile = initiativeFile(boardId, initiativeId); - - if (!existsSync(dataFile) && !existsSync(dir)) return false; - - if (existsSync(dataFile)) await rm(dataFile); - if (existsSync(dir)) await rm(dir, { recursive: true }); - - await this.appendActivity(boardId, { - type: "initiative.removed", - message: `Initiative ${initiativeId} removed`, - initiativeId, - }); - - return true; - } - - // --- Plans --- - - async createPlan( - boardId: string, - initiativeId: string, - data: Pick & - Partial>, - ): Promise { - await this.ensureBoardStructure(boardId); - - const initiative = await this.getInitiative(boardId, initiativeId); - if (!initiative) { - throw new Error(`Initiative not found: ${initiativeId}`); - } - - const baseId = slugify(data.title); - const id = await resolveUniqueId((candidate) => planFile(boardId, initiativeId, candidate), baseId); - const now = timestamp(); - - const plan: Plan = { - id, - boardId, - initiativeId, - title: data.title, - description: data.description || "", - status: data.status || "todo", - ownerAgentId: data.ownerAgentId || null, - tags: data.tags || [], - createdAt: now, - updatedAt: now, - completedAt: null, - }; - - await writeJson(planFile(boardId, initiativeId, id), plan); - await mkdir(planStepsDir(boardId, initiativeId, id), { recursive: true }); - - await this.appendActivity(boardId, { - type: "plan.created", - message: `Plan ${plan.title} created`, - initiativeId, - payload: { planId: plan.id, status: plan.status }, - agentId: plan.ownerAgentId || undefined, - }); - - return plan; - } - - async listPlans(boardId: string, initiativeId: string): Promise { - await this.ensureBoardStructure(boardId); - - const files = await listDir(plansDir(boardId, initiativeId)); - const plans: Plan[] = []; - for (const fileName of files) { - if (!fileName.endsWith(".json")) continue; - const plan = await readJson(path.join(plansDir(boardId, initiativeId), fileName), isEntityRecord); - if (!plan) continue; - plans.push({ - ...plan, - status: plan.status || "todo", - ownerAgentId: plan.ownerAgentId || null, - tags: plan.tags || [], - completedAt: plan.completedAt || null, - }); - } - - return plans.sort((a, b) => compareStr(a.createdAt, b.createdAt)); - } - - async getPlan(boardId: string, initiativeId: string, planId: string): Promise { - await this.ensureBoardStructure(boardId); - - const plan = await readJson(planFile(boardId, initiativeId, planId), isEntityRecord); - if (!plan) return null; - - return { - ...plan, - status: plan.status || "todo", - ownerAgentId: plan.ownerAgentId || null, - tags: plan.tags || [], - completedAt: plan.completedAt || null, - }; - } - - async updatePlan( - boardId: string, - initiativeId: string, - planId: string, - data: Partial>, - ): Promise { - await this.ensureBoardStructure(boardId); - - const current = await this.getPlan(boardId, initiativeId, planId); - if (!current) return null; - - const now = timestamp(); - const nextStatus = data.status || current.status; - const completedAt = - nextStatus === "done" ? now : data.status && data.status !== "done" ? null : current.completedAt; - - const updated: Plan = { - ...current, - ...data, - status: nextStatus, - updatedAt: now, - completedAt, - ownerAgentId: data.ownerAgentId === undefined ? current.ownerAgentId : data.ownerAgentId, - tags: data.tags || current.tags, - }; - - await writeJson(planFile(boardId, initiativeId, planId), updated); - - const statusChange = data.status && data.status !== current.status; - await this.appendActivity(boardId, { - type: statusChange ? "plan.status_changed" : "plan.updated", - message: statusChange - ? `Plan ${updated.title} moved ${current.status} -> ${updated.status}` - : `Plan ${updated.title} updated`, - initiativeId, - payload: { planId, status: updated.status, fields: Object.keys(data) }, - agentId: updated.ownerAgentId || undefined, - }); - - return updated; - } - - async deletePlan(boardId: string, initiativeId: string, planId: string): Promise { - await this.ensureBoardStructure(boardId); - - const filePath = planFile(boardId, initiativeId, planId); - const dirPath = path.join(plansDir(boardId, initiativeId), planId); - if (!existsSync(filePath) && !existsSync(dirPath)) return false; - - const tasks = await this.listTasks(boardId, initiativeId); - for (const task of tasks) { - if (task.planId !== planId) continue; - const detached: Task = { ...task, planId: null, planStepId: null, updatedAt: timestamp() }; - await writeJson(taskFile(boardId, initiativeId, task.id), detached); - } - - if (existsSync(filePath)) await rm(filePath); - if (existsSync(dirPath)) await rm(dirPath, { recursive: true }); - - await this.appendActivity(boardId, { - type: "plan.removed", - message: `Plan ${planId} removed`, - initiativeId, - payload: { planId }, - }); - - return true; - } - - // --- Plan steps --- - - async createPlanStep( - boardId: string, - initiativeId: string, - planId: string, - data: Pick & - Partial>, - ): Promise { - await this.ensureBoardStructure(boardId); - - const plan = await this.getPlan(boardId, initiativeId, planId); - if (!plan) { - throw new Error(`Plan not found: ${planId}`); - } - - const baseId = slugify(data.title); - const id = await resolveUniqueId( - (candidate) => planStepFile(boardId, initiativeId, planId, candidate), - baseId, - ); - const now = timestamp(); - const existingSteps = await this.listPlanSteps(boardId, initiativeId, planId); - const order = data.order ?? existingSteps.length + 1; - - const step: PlanStep = { - id, - boardId, - initiativeId, - planId, - title: data.title, - description: data.description || "", - status: data.status || "todo", - assigneeAgentId: data.assigneeAgentId || null, - order, - createdAt: now, - updatedAt: now, - completedAt: null, - }; - - await writeJson(planStepFile(boardId, initiativeId, planId, id), step); - - await this.appendActivity(boardId, { - type: "plan_step.created", - message: `Plan step ${step.title} created`, - initiativeId, - payload: { planId, stepId: step.id, status: step.status, order: step.order }, - agentId: step.assigneeAgentId || undefined, - }); - - return step; - } - - async listPlanSteps(boardId: string, initiativeId: string, planId: string): Promise { - await this.ensureBoardStructure(boardId); - - const files = await listDir(planStepsDir(boardId, initiativeId, planId)); - const steps: PlanStep[] = []; - for (const fileName of files) { - if (!fileName.endsWith(".json")) continue; - const step = await readJson( - path.join(planStepsDir(boardId, initiativeId, planId), fileName), - isEntityRecord, - ); - if (!step) continue; - steps.push({ - ...step, - status: step.status || "todo", - assigneeAgentId: step.assigneeAgentId || null, - completedAt: step.completedAt || null, - }); - } - - return steps.sort((a, b) => (a.order ?? 0) - (b.order ?? 0) || compareStr(a.createdAt, b.createdAt)); - } - - async getPlanStep( - boardId: string, - initiativeId: string, - planId: string, - stepId: string, - ): Promise { - await this.ensureBoardStructure(boardId); - - const step = await readJson(planStepFile(boardId, initiativeId, planId, stepId), isEntityRecord); - if (!step) return null; - - return { - ...step, - status: step.status || "todo", - assigneeAgentId: step.assigneeAgentId || null, - completedAt: step.completedAt || null, - }; - } - - async updatePlanStep( - boardId: string, - initiativeId: string, - planId: string, - stepId: string, - data: Partial>, - ): Promise { - await this.ensureBoardStructure(boardId); - - const current = await this.getPlanStep(boardId, initiativeId, planId, stepId); - if (!current) return null; - - const now = timestamp(); - const nextStatus = data.status || current.status; - const completedAt = - nextStatus === "done" ? now : data.status && data.status !== "done" ? null : current.completedAt; - - const updated: PlanStep = { - ...current, - ...data, - status: nextStatus, - updatedAt: now, - completedAt, - assigneeAgentId: - data.assigneeAgentId === undefined ? current.assigneeAgentId : data.assigneeAgentId, - order: data.order ?? current.order, - }; - - await writeJson(planStepFile(boardId, initiativeId, planId, stepId), updated); - - const statusChange = data.status && data.status !== current.status; - await this.appendActivity(boardId, { - type: statusChange ? "plan_step.status_changed" : "plan_step.updated", - message: statusChange - ? `Plan step ${updated.title} moved ${current.status} -> ${updated.status}` - : `Plan step ${updated.title} updated`, - initiativeId, - payload: { planId, stepId, status: updated.status, fields: Object.keys(data) }, - agentId: updated.assigneeAgentId || undefined, - }); - - return updated; - } - - async deletePlanStep(boardId: string, initiativeId: string, planId: string, stepId: string): Promise { - await this.ensureBoardStructure(boardId); - - const filePath = planStepFile(boardId, initiativeId, planId, stepId); - if (!existsSync(filePath)) return false; - - // Detach tasks FIRST, then delete the step file — prevents orphaned task references on crash - const tasks = await this.listTasks(boardId, initiativeId); - for (const task of tasks) { - if (task.planId === planId && task.planStepId === stepId) { - const detached: Task = { ...task, planStepId: null, updatedAt: timestamp() }; - await writeJson(taskFile(boardId, initiativeId, task.id), detached); - } - } - - await rm(filePath); - - await this.appendActivity(boardId, { - type: "plan_step.removed", - message: `Plan step ${stepId} removed`, - initiativeId, - payload: { planId, stepId }, - }); - - return true; - } - - // --- Projects (legacy compatibility wrappers) --- - - async createProject(boardId: string, data: Pick): Promise { - return this.createInitiative(boardId, data); - } - - async listProjects(boardId: string): Promise { - return this.listInitiatives(boardId); - } - - async getProject(boardId: string, projectId: string): Promise { - return this.getInitiative(boardId, projectId); - } - - async updateProject( - boardId: string, - projectId: string, - data: Partial>, - ): Promise { - return this.updateInitiative(boardId, projectId, data); - } - - async deleteProject(boardId: string, projectId: string): Promise { - return this.deleteInitiative(boardId, projectId); - } - - // --- Tasks --- - - async createTask( - boardId: string, - initiativeId: string, - data: Pick & - Partial>, - actorAgentId: string, - ): Promise { - await this.ensureBoardStructure(boardId); - await this.requireRegisteredAgent(boardId, actorAgentId); - - const initiative = await this.getInitiative(boardId, initiativeId); - if (!initiative) { - throw new Error(`Initiative not found: ${initiativeId}`); - } - if (data.planId) { - const plan = await this.getPlan(boardId, initiativeId, data.planId); - if (!plan) { - throw new Error(`Plan not found: ${data.planId}`); - } - if (data.planStepId) { - const step = await this.getPlanStep(boardId, initiativeId, data.planId, data.planStepId); - if (!step) { - throw new Error(`Plan step not found: ${data.planStepId}`); - } - } - } else if (data.planStepId) { - throw new Error("planStepId requires a planId"); - } - - const baseId = slugify(data.title); - const id = await resolveUniqueId((candidate) => taskFile(boardId, initiativeId, candidate), baseId); - const now = timestamp(); - - const task = normalizeTask( - { - id, - initiativeId, - projectId: initiativeId, - planId: data.planId || null, - planStepId: data.planStepId || null, - boardId, - title: data.title, - description: data.description || "", - status: "todo", - assigneeAgentId: data.assigneeAgentId || null, - assigneeAgentIds: data.assigneeAgentIds || [], - deliverables: data.deliverables || [], - priority: data.priority || "medium", - tags: data.tags || [], - createdAt: now, - updatedAt: now, - completedAt: null, - }, - initiativeId, - ); - - await writeJson(taskFile(boardId, initiativeId, id), task); - - await this.appendActivity(boardId, { - type: "task.created", - message: `Task ${task.title} created`, - initiativeId, - taskId: task.id, - agentId: actorAgentId, - payload: { - assignees: task.assigneeAgentIds, - status: task.status, - }, - }); - - return task; - } - - async listTasks(boardId: string, initiativeId: string, filters?: { status?: string; assignee?: string; tag?: string }): Promise { - await this.ensureBoardStructure(boardId); - - const files = await listDir(tasksDir(boardId, initiativeId)); - let tasks: Task[] = []; - - for (const f of files) { - if (!f.endsWith(".json")) continue; - const task = await readJson(path.join(tasksDir(boardId, initiativeId), f), isEntityRecord); - if (task) { - tasks.push(normalizeTask(task, initiativeId)); - } - } - - if (filters?.status) tasks = tasks.filter((t) => t.status === filters.status); - if (filters?.assignee) { - tasks = tasks.filter( - (t) => t.assigneeAgentId === filters.assignee || t.assigneeAgentIds.includes(filters.assignee as string), - ); - } - if (filters?.tag) tasks = tasks.filter((t) => t.tags.includes(filters.tag as string)); - - return tasks.sort((a, b) => compareStr(a.createdAt, b.createdAt)); - } - - async listAllBoardTasks(boardId: string, filters?: { status?: string; assignee?: string; tag?: string }): Promise { - const initiatives = await this.listInitiatives(boardId); - const allTasks: Task[] = []; - - for (const initiative of initiatives) { - const tasks = await this.listTasks(boardId, initiative.id, filters); - allTasks.push(...tasks); - } - - return allTasks.sort((a, b) => compareStr(b.updatedAt, a.updatedAt)); - } - - async getTask(boardId: string, initiativeId: string, taskId: string): Promise { - await this.ensureBoardStructure(boardId); - - const task = await readJson(taskFile(boardId, initiativeId, taskId), isEntityRecord); - if (!task) return null; - - return normalizeTask(task, initiativeId); - } - - async updateTask( - boardId: string, - initiativeId: string, - taskId: string, - data: Partial< - Pick< - Task, - "title" | "description" | "status" | "planId" | "planStepId" | "assigneeAgentId" | "assigneeAgentIds" | "deliverables" | "priority" | "tags" - > - >, - actorAgentId: string, - ): Promise { - await this.ensureBoardStructure(boardId); - await this.requireRegisteredAgent(boardId, actorAgentId); - - const current = await this.getTask(boardId, initiativeId, taskId); - if (!current) return null; - - const nextPlanId = data.planId === undefined ? current.planId : data.planId; - const nextPlanStepId = data.planStepId === undefined ? current.planStepId : data.planStepId; - - if (nextPlanId) { - const plan = await this.getPlan(boardId, initiativeId, nextPlanId); - if (!plan) { - throw new Error(`Plan not found: ${nextPlanId}`); - } - if (nextPlanStepId) { - const step = await this.getPlanStep(boardId, initiativeId, nextPlanId, nextPlanStepId); - if (!step) { - throw new Error(`Plan step not found: ${nextPlanStepId}`); - } - } - } else if (nextPlanStepId) { - throw new Error("planStepId requires a planId"); - } - - const now = timestamp(); - const nextStatus = data.status || current.status; - const completedAt = - nextStatus === "done" ? now : data.status && data.status !== "done" ? null : current.completedAt; - - const updated = normalizeTask( - { - ...current, - ...data, - status: nextStatus, - updatedAt: now, - completedAt, - }, - initiativeId, - ); - - await writeJson(taskFile(boardId, initiativeId, taskId), updated); - await this.syncAgentPointersForTask(boardId, updated); - - const changedFields = Object.keys(data); - const statusChange = data.status && data.status !== current.status; - - await this.appendActivity(boardId, { - type: statusChange ? "task.status_changed" : "task.updated", - message: statusChange - ? `Task ${updated.title} moved ${current.status} -> ${updated.status}` - : `Task ${updated.title} updated`, - initiativeId, - taskId, - agentId: actorAgentId, - payload: { fields: changedFields, status: updated.status }, - }); - - return updated; - } - - async deleteTask(boardId: string, initiativeId: string, taskId: string, actorAgentId: string): Promise { - await this.ensureBoardStructure(boardId); - await this.requireRegisteredAgent(boardId, actorAgentId); - - const filePath = taskFile(boardId, initiativeId, taskId); - if (!existsSync(filePath)) return false; - - const task = await this.getTask(boardId, initiativeId, taskId); - await rm(filePath, { force: true }); - - if (task) { - await this.syncAgentPointersForTask(boardId, { ...task, status: "todo" as TaskStatus }); - await this.appendActivity(boardId, { - type: "task.removed", - message: `Task ${task.title} removed`, - initiativeId, - taskId, - agentId: actorAgentId, - }); - } - - return true; - } - - // --- Activity --- - - async listActivity( - boardId: string, - options?: { limit?: number; initiativeId?: string; agentId?: string; taskId?: string }, - ): Promise { - await this.ensureBoardStructure(boardId); - - const files = (await listDir(eventsDir(boardId))) - .filter((name) => name.endsWith(".ndjson")) - .sort((a, b) => b.localeCompare(a)); - - const limit = options?.limit ?? 200; - const hasFilters = !!(options?.initiativeId || options?.agentId || options?.taskId); - const events: ActivityEvent[] = []; - - for (const fileName of files) { - // Files are sorted descending by date — stop early when no filters are active - if (!hasFilters && events.length >= limit) break; - - const filePath = path.join(eventsDir(boardId), fileName); - const content = await readFile(filePath, "utf-8").catch(() => ""); - if (!content) continue; - - const lines = content.split("\n").filter(Boolean); - for (const line of lines) { - try { - const event = JSON.parse(line) as ActivityEvent; - events.push(event); - } catch { - // Skip malformed lines and continue. - } - } - } - - let filtered = events; - if (options?.initiativeId) { - filtered = filtered.filter((e) => e.initiativeId === options.initiativeId); - } - if (options?.agentId) { - filtered = filtered.filter((e) => e.agentId === options.agentId); - } - if (options?.taskId) { - filtered = filtered.filter((e) => e.taskId === options.taskId); - } - - filtered.sort((a, b) => compareStr(b.createdAt, a.createdAt)); - - return filtered.slice(0, limit); - } -} - -// Singleton — use globalThis to survive Next.js HMR in development -const globalForStorage = globalThis as unknown as { _agentboardStorage?: FsStorage }; - -export function getStorage(): Storage { - if (!globalForStorage._agentboardStorage) { - globalForStorage._agentboardStorage = new FsStorage(); } - return globalForStorage._agentboardStorage; + return globalForStorage._agentboardStoragePromise; } diff --git a/src/lib/storage/memory-object-store.ts b/src/lib/storage/memory-object-store.ts new file mode 100644 index 0000000..bf2bf23 --- /dev/null +++ b/src/lib/storage/memory-object-store.ts @@ -0,0 +1,81 @@ +import type { ObjectStore } from "./object-store"; + +/** + * In-memory ObjectStore. Not for production — used to exercise DocumentStorage's + * business logic independently of any real backend. + */ +export class InMemoryObjectStore implements ObjectStore { + private store = new Map(); + + async get(key: string, validate?: (value: unknown) => boolean): Promise { + const raw = this.store.get(key); + if (raw === undefined) return null; + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + this.store.delete(key); + return null; + } + if (validate && !validate(parsed)) return null; + return parsed as T; + } + + async put(key: string, data: unknown): Promise { + this.store.set(key, JSON.stringify(data)); + } + + async createIfAbsent(key: string): Promise { + if (this.store.has(key)) return false; + this.store.set(key, "{}"); + return true; + } + + async exists(key: string): Promise { + return this.store.has(key); + } + + async delete(key: string): Promise { + this.store.delete(key); + } + + async deletePrefix(prefix: string): Promise { + const p = prefix.endsWith("/") ? prefix : `${prefix}/`; + for (const key of [...this.store.keys()]) { + if (key.startsWith(p)) this.store.delete(key); + } + } + + async listChildren(prefix: string): Promise { + const p = prefix.endsWith("/") ? prefix : `${prefix}/`; + const names = new Set(); + for (const key of this.store.keys()) { + if (!key.startsWith(p)) continue; + const segment = key.slice(p.length).split("/")[0]; + if (segment) names.add(segment); + } + return [...names]; + } + + async appendEvent(eventsKey: string, bucket: string, record: unknown): Promise { + this.store.set(`${eventsKey}/${bucket}/${this.store.size}-${Math.random().toString(36).slice(2)}`, JSON.stringify(record)); + } + + async listEventBuckets(eventsKey: string): Promise { + return this.listChildren(eventsKey); + } + + async readEventBucket(eventsKey: string, bucket: string): Promise { + const p = `${eventsKey}/${bucket}/`; + const records: unknown[] = []; + for (const [key, raw] of this.store.entries()) { + if (!key.startsWith(p)) continue; + try { + records.push(JSON.parse(raw)); + } catch { + // skip + } + } + return records; + } +} diff --git a/src/lib/storage/object-store.ts b/src/lib/storage/object-store.ts new file mode 100644 index 0000000..1531780 --- /dev/null +++ b/src/lib/storage/object-store.ts @@ -0,0 +1,56 @@ +/** + * Low-level key/value primitives that back the high-level document storage. + * + * Keys are POSIX-style relative paths (e.g. `boards/foo/agents/bar.json`). + * The filesystem backend maps them under the data dir; the Vercel Blob backend + * maps them to blob pathnames. All business logic lives in DocumentStorage and + * speaks only this interface, so swapping the backend (local FS vs. serverless + * Blob) requires no changes to the storage semantics. + */ +export interface ObjectStore { + /** + * Reads and parses the JSON value at `key`. + * - Missing key -> null. + * - Unparseable content -> null (backend may quarantine it). + * - When `validate` is supplied and returns false -> null (value skipped). + */ + get(key: string, validate?: (value: unknown) => boolean): Promise; + + /** Writes `data` as JSON at `key`, creating any parent containers. */ + put(key: string, data: unknown): Promise; + + /** + * Atomically creates a placeholder at `key` only if it does not already + * exist. Returns true if it was created, false if the key was already taken. + * Used to reserve unique ids without a TOCTOU race. + */ + createIfAbsent(key: string): Promise; + + /** Whether a value exists at `key`. */ + exists(key: string): Promise; + + /** Deletes the value at `key`. No-op if it does not exist. */ + delete(key: string): Promise; + + /** Recursively deletes every key under `prefix`. No-op if empty. */ + deletePrefix(prefix: string): Promise; + + /** + * Lists the direct child entry names under the directory `prefix` — both + * leaf files (e.g. `bar.json`) and subdirectory names (e.g. `bar`), matching + * `fs.readdir` semantics. Does not recurse. + */ + listChildren(prefix: string): Promise; + + /** + * Appends an event record to an append-only collection rooted at `eventsKey`, + * partitioned into `bucket` (e.g. a date). Concurrent appends are safe. + */ + appendEvent(eventsKey: string, bucket: string, record: unknown): Promise; + + /** Lists the bucket names under an events collection. */ + listEventBuckets(eventsKey: string): Promise; + + /** Reads every event record in a bucket, skipping unparseable ones. */ + readEventBucket(eventsKey: string, bucket: string): Promise; +}