mirror of
https://github.com/snapotter-hq/SnapOtter.git
synced 2026-08-03 07:46:42 +02:00
feat(api): add pipeline execution, save, and list endpoints
Add pipelines table to SQLite schema with Drizzle migration. Implement POST /api/v1/pipeline/execute (sequential multi-tool processing), POST /api/v1/pipeline/save, GET /api/v1/pipeline/list, DELETE /api/v1/pipeline/:id. Pipeline execution validates all tool IDs and settings before processing, chains output of each step as input to the next.
This commit is contained in:
Siddharth Kumar Sah
@@ -0,0 +1,7 @@
|
|||||||
|
CREATE TABLE `pipelines` (
|
||||||
|
`id` text PRIMARY KEY NOT NULL,
|
||||||
|
`name` text NOT NULL,
|
||||||
|
`description` text,
|
||||||
|
`steps` text NOT NULL,
|
||||||
|
`created_at` integer NOT NULL
|
||||||
|
);
|
||||||
@@ -0,0 +1,364 @@
|
|||||||
|
{
|
||||||
|
"version": "6",
|
||||||
|
"dialect": "sqlite",
|
||||||
|
"id": "91a14a95-bbcb-46ef-abe3-6d2f6fbc8458",
|
||||||
|
"prevId": "c7909605-aabf-4832-8ef8-9390c0f7c99a",
|
||||||
|
"tables": {
|
||||||
|
"api_keys": {
|
||||||
|
"name": "api_keys",
|
||||||
|
"columns": {
|
||||||
|
"id": {
|
||||||
|
"name": "id",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": true,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"user_id": {
|
||||||
|
"name": "user_id",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"key_hash": {
|
||||||
|
"name": "key_hash",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"name": {
|
||||||
|
"name": "name",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false,
|
||||||
|
"default": "'Default API Key'"
|
||||||
|
},
|
||||||
|
"created_at": {
|
||||||
|
"name": "created_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"last_used_at": {
|
||||||
|
"name": "last_used_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": false,
|
||||||
|
"autoincrement": false
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"indexes": {},
|
||||||
|
"foreignKeys": {
|
||||||
|
"api_keys_user_id_users_id_fk": {
|
||||||
|
"name": "api_keys_user_id_users_id_fk",
|
||||||
|
"tableFrom": "api_keys",
|
||||||
|
"tableTo": "users",
|
||||||
|
"columnsFrom": [
|
||||||
|
"user_id"
|
||||||
|
],
|
||||||
|
"columnsTo": [
|
||||||
|
"id"
|
||||||
|
],
|
||||||
|
"onDelete": "cascade",
|
||||||
|
"onUpdate": "no action"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"compositePrimaryKeys": {},
|
||||||
|
"uniqueConstraints": {},
|
||||||
|
"checkConstraints": {}
|
||||||
|
},
|
||||||
|
"jobs": {
|
||||||
|
"name": "jobs",
|
||||||
|
"columns": {
|
||||||
|
"id": {
|
||||||
|
"name": "id",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": true,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"type": {
|
||||||
|
"name": "type",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"status": {
|
||||||
|
"name": "status",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false,
|
||||||
|
"default": "'queued'"
|
||||||
|
},
|
||||||
|
"progress": {
|
||||||
|
"name": "progress",
|
||||||
|
"type": "real",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false,
|
||||||
|
"default": 0
|
||||||
|
},
|
||||||
|
"input_files": {
|
||||||
|
"name": "input_files",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"output_path": {
|
||||||
|
"name": "output_path",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": false,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"settings": {
|
||||||
|
"name": "settings",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": false,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"error": {
|
||||||
|
"name": "error",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": false,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"created_at": {
|
||||||
|
"name": "created_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"completed_at": {
|
||||||
|
"name": "completed_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": false,
|
||||||
|
"autoincrement": false
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"indexes": {},
|
||||||
|
"foreignKeys": {},
|
||||||
|
"compositePrimaryKeys": {},
|
||||||
|
"uniqueConstraints": {},
|
||||||
|
"checkConstraints": {}
|
||||||
|
},
|
||||||
|
"pipelines": {
|
||||||
|
"name": "pipelines",
|
||||||
|
"columns": {
|
||||||
|
"id": {
|
||||||
|
"name": "id",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": true,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"name": {
|
||||||
|
"name": "name",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"description": {
|
||||||
|
"name": "description",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": false,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"steps": {
|
||||||
|
"name": "steps",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"created_at": {
|
||||||
|
"name": "created_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"indexes": {},
|
||||||
|
"foreignKeys": {},
|
||||||
|
"compositePrimaryKeys": {},
|
||||||
|
"uniqueConstraints": {},
|
||||||
|
"checkConstraints": {}
|
||||||
|
},
|
||||||
|
"sessions": {
|
||||||
|
"name": "sessions",
|
||||||
|
"columns": {
|
||||||
|
"id": {
|
||||||
|
"name": "id",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": true,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"user_id": {
|
||||||
|
"name": "user_id",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"expires_at": {
|
||||||
|
"name": "expires_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"created_at": {
|
||||||
|
"name": "created_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"indexes": {},
|
||||||
|
"foreignKeys": {
|
||||||
|
"sessions_user_id_users_id_fk": {
|
||||||
|
"name": "sessions_user_id_users_id_fk",
|
||||||
|
"tableFrom": "sessions",
|
||||||
|
"tableTo": "users",
|
||||||
|
"columnsFrom": [
|
||||||
|
"user_id"
|
||||||
|
],
|
||||||
|
"columnsTo": [
|
||||||
|
"id"
|
||||||
|
],
|
||||||
|
"onDelete": "cascade",
|
||||||
|
"onUpdate": "no action"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"compositePrimaryKeys": {},
|
||||||
|
"uniqueConstraints": {},
|
||||||
|
"checkConstraints": {}
|
||||||
|
},
|
||||||
|
"settings": {
|
||||||
|
"name": "settings",
|
||||||
|
"columns": {
|
||||||
|
"key": {
|
||||||
|
"name": "key",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": true,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"value": {
|
||||||
|
"name": "value",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"updated_at": {
|
||||||
|
"name": "updated_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"indexes": {},
|
||||||
|
"foreignKeys": {},
|
||||||
|
"compositePrimaryKeys": {},
|
||||||
|
"uniqueConstraints": {},
|
||||||
|
"checkConstraints": {}
|
||||||
|
},
|
||||||
|
"users": {
|
||||||
|
"name": "users",
|
||||||
|
"columns": {
|
||||||
|
"id": {
|
||||||
|
"name": "id",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": true,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"username": {
|
||||||
|
"name": "username",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"password_hash": {
|
||||||
|
"name": "password_hash",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"role": {
|
||||||
|
"name": "role",
|
||||||
|
"type": "text",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false,
|
||||||
|
"default": "'user'"
|
||||||
|
},
|
||||||
|
"must_change_password": {
|
||||||
|
"name": "must_change_password",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false,
|
||||||
|
"default": true
|
||||||
|
},
|
||||||
|
"created_at": {
|
||||||
|
"name": "created_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
},
|
||||||
|
"updated_at": {
|
||||||
|
"name": "updated_at",
|
||||||
|
"type": "integer",
|
||||||
|
"primaryKey": false,
|
||||||
|
"notNull": true,
|
||||||
|
"autoincrement": false
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"indexes": {
|
||||||
|
"users_username_unique": {
|
||||||
|
"name": "users_username_unique",
|
||||||
|
"columns": [
|
||||||
|
"username"
|
||||||
|
],
|
||||||
|
"isUnique": true
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"foreignKeys": {},
|
||||||
|
"compositePrimaryKeys": {},
|
||||||
|
"uniqueConstraints": {},
|
||||||
|
"checkConstraints": {}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"views": {},
|
||||||
|
"enums": {},
|
||||||
|
"_meta": {
|
||||||
|
"schemas": {},
|
||||||
|
"tables": {},
|
||||||
|
"columns": {}
|
||||||
|
},
|
||||||
|
"internal": {
|
||||||
|
"indexes": {}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -8,6 +8,13 @@
|
|||||||
"when": 1774119039901,
|
"when": 1774119039901,
|
||||||
"tag": "0000_clammy_madelyne_pryor",
|
"tag": "0000_clammy_madelyne_pryor",
|
||||||
"breakpoints": true
|
"breakpoints": true
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"idx": 1,
|
||||||
|
"version": "6",
|
||||||
|
"when": 1774125684700,
|
||||||
|
"tag": "0001_amusing_omega_red",
|
||||||
|
"breakpoints": true
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
@@ -44,3 +44,11 @@ export const apiKeys = sqliteTable("api_keys", {
|
|||||||
createdAt: integer("created_at", { mode: "timestamp" }).notNull().$defaultFn(() => new Date()),
|
createdAt: integer("created_at", { mode: "timestamp" }).notNull().$defaultFn(() => new Date()),
|
||||||
lastUsedAt: integer("last_used_at", { mode: "timestamp" }),
|
lastUsedAt: integer("last_used_at", { mode: "timestamp" }),
|
||||||
});
|
});
|
||||||
|
|
||||||
|
export const pipelines = sqliteTable("pipelines", {
|
||||||
|
id: text("id").primaryKey(),
|
||||||
|
name: text("name").notNull(),
|
||||||
|
description: text("description"),
|
||||||
|
steps: text("steps").notNull(), // JSON array of { toolId, settings }
|
||||||
|
createdAt: integer("created_at", { mode: "timestamp" }).notNull().$defaultFn(() => new Date()),
|
||||||
|
});
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import { startCleanupCron } from "./lib/cleanup.js";
|
|||||||
import { fileRoutes } from "./routes/files.js";
|
import { fileRoutes } from "./routes/files.js";
|
||||||
import { registerToolRoutes } from "./routes/tools/index.js";
|
import { registerToolRoutes } from "./routes/tools/index.js";
|
||||||
import { registerBatchRoutes } from "./routes/batch.js";
|
import { registerBatchRoutes } from "./routes/batch.js";
|
||||||
|
import { registerPipelineRoutes } from "./routes/pipeline.js";
|
||||||
import { registerProgressRoutes } from "./routes/progress.js";
|
import { registerProgressRoutes } from "./routes/progress.js";
|
||||||
|
|
||||||
// Run before anything else
|
// Run before anything else
|
||||||
@@ -68,6 +69,9 @@ await registerToolRoutes(app);
|
|||||||
// Batch processing routes (must be after tool routes so the registry is populated)
|
// Batch processing routes (must be after tool routes so the registry is populated)
|
||||||
await registerBatchRoutes(app);
|
await registerBatchRoutes(app);
|
||||||
|
|
||||||
|
// Pipeline routes (must be after tool routes so the registry is populated)
|
||||||
|
await registerPipelineRoutes(app);
|
||||||
|
|
||||||
// Progress SSE routes
|
// Progress SSE routes
|
||||||
await registerProgressRoutes(app);
|
await registerProgressRoutes(app);
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,312 @@
|
|||||||
|
/**
|
||||||
|
* Pipeline execution, save, list, and delete routes.
|
||||||
|
*
|
||||||
|
* POST /api/v1/pipeline/execute — Execute a pipeline (array of tool steps)
|
||||||
|
* POST /api/v1/pipeline/save — Save a pipeline definition
|
||||||
|
* GET /api/v1/pipeline/list — List saved pipelines
|
||||||
|
* DELETE /api/v1/pipeline/:id — Delete a saved pipeline
|
||||||
|
*/
|
||||||
|
import { randomUUID } from "node:crypto";
|
||||||
|
import { writeFile } from "node:fs/promises";
|
||||||
|
import { join, basename } from "node:path";
|
||||||
|
import type { FastifyInstance, FastifyRequest, FastifyReply } from "fastify";
|
||||||
|
import { eq } from "drizzle-orm";
|
||||||
|
import { z } from "zod";
|
||||||
|
import { getToolConfig } from "./tool-factory.js";
|
||||||
|
import { validateImageBuffer } from "../lib/file-validation.js";
|
||||||
|
import { createWorkspace } from "../lib/workspace.js";
|
||||||
|
import { db, schema } from "../db/index.js";
|
||||||
|
|
||||||
|
/** Schema for a single pipeline step. */
|
||||||
|
const pipelineStepSchema = z.object({
|
||||||
|
toolId: z.string(),
|
||||||
|
settings: z.record(z.unknown()).default({}),
|
||||||
|
});
|
||||||
|
|
||||||
|
/** Schema for a full pipeline definition. */
|
||||||
|
const pipelineDefinitionSchema = z.object({
|
||||||
|
steps: z.array(pipelineStepSchema).min(1, "Pipeline must have at least one step"),
|
||||||
|
});
|
||||||
|
|
||||||
|
/** Schema for saving a pipeline. */
|
||||||
|
const savePipelineSchema = z.object({
|
||||||
|
name: z.string().min(1, "Pipeline name is required").max(100),
|
||||||
|
description: z.string().max(500).optional(),
|
||||||
|
steps: z.array(pipelineStepSchema).min(1, "Pipeline must have at least one step"),
|
||||||
|
});
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Sanitize a filename to prevent path traversal attacks.
|
||||||
|
*/
|
||||||
|
function sanitizeFilename(raw: string): string {
|
||||||
|
let name = basename(raw);
|
||||||
|
name = name.replace(/\.\./g, "");
|
||||||
|
name = name.replace(/\0/g, "");
|
||||||
|
if (!name || name === "." || name === "..") {
|
||||||
|
name = "image";
|
||||||
|
}
|
||||||
|
return name;
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function registerPipelineRoutes(app: FastifyInstance): Promise<void> {
|
||||||
|
/**
|
||||||
|
* POST /api/v1/pipeline/execute
|
||||||
|
*
|
||||||
|
* Accepts multipart with:
|
||||||
|
* - A file part (the image to process)
|
||||||
|
* - A "pipeline" field containing JSON: { steps: [{ toolId, settings }, ...] }
|
||||||
|
*
|
||||||
|
* Processes the image through each step sequentially.
|
||||||
|
* The output of step N becomes the input of step N+1.
|
||||||
|
* Returns the final processed image for download.
|
||||||
|
*/
|
||||||
|
app.post(
|
||||||
|
"/api/v1/pipeline/execute",
|
||||||
|
async (request: FastifyRequest, reply: FastifyReply) => {
|
||||||
|
let fileBuffer: Buffer | null = null;
|
||||||
|
let filename = "image";
|
||||||
|
let pipelineRaw: string | null = null;
|
||||||
|
|
||||||
|
// Parse multipart
|
||||||
|
try {
|
||||||
|
const parts = request.parts();
|
||||||
|
for await (const part of parts) {
|
||||||
|
if (part.type === "file") {
|
||||||
|
const chunks: Buffer[] = [];
|
||||||
|
for await (const chunk of part.file) {
|
||||||
|
chunks.push(chunk);
|
||||||
|
}
|
||||||
|
fileBuffer = Buffer.concat(chunks);
|
||||||
|
filename = sanitizeFilename(part.filename ?? "image");
|
||||||
|
} else if (part.fieldname === "pipeline") {
|
||||||
|
pipelineRaw = part.value as string;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
return reply.status(400).send({
|
||||||
|
error: "Failed to parse multipart request",
|
||||||
|
details: err instanceof Error ? err.message : String(err),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!fileBuffer || fileBuffer.length === 0) {
|
||||||
|
return reply.status(400).send({ error: "No image file provided" });
|
||||||
|
}
|
||||||
|
|
||||||
|
// Validate the initial image
|
||||||
|
const validation = await validateImageBuffer(fileBuffer);
|
||||||
|
if (!validation.valid) {
|
||||||
|
return reply.status(400).send({
|
||||||
|
error: `Invalid image: ${validation.reason}`,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
// Parse and validate the pipeline definition
|
||||||
|
if (!pipelineRaw) {
|
||||||
|
return reply.status(400).send({ error: "No pipeline definition provided" });
|
||||||
|
}
|
||||||
|
|
||||||
|
let pipeline: z.infer<typeof pipelineDefinitionSchema>;
|
||||||
|
try {
|
||||||
|
const parsed = JSON.parse(pipelineRaw);
|
||||||
|
const result = pipelineDefinitionSchema.safeParse(parsed);
|
||||||
|
if (!result.success) {
|
||||||
|
return reply.status(400).send({
|
||||||
|
error: "Invalid pipeline definition",
|
||||||
|
details: result.error.issues.map((i) => ({
|
||||||
|
path: i.path.join("."),
|
||||||
|
message: i.message,
|
||||||
|
})),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
pipeline = result.data;
|
||||||
|
} catch {
|
||||||
|
return reply.status(400).send({ error: "Pipeline must be valid JSON" });
|
||||||
|
}
|
||||||
|
|
||||||
|
// Validate all tool IDs exist before starting
|
||||||
|
for (let i = 0; i < pipeline.steps.length; i++) {
|
||||||
|
const step = pipeline.steps[i];
|
||||||
|
const toolConfig = getToolConfig(step.toolId);
|
||||||
|
if (!toolConfig) {
|
||||||
|
return reply.status(400).send({
|
||||||
|
error: `Step ${i + 1}: Tool "${step.toolId}" not found`,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
// Validate the settings for this tool
|
||||||
|
const settingsResult = toolConfig.settingsSchema.safeParse(step.settings);
|
||||||
|
if (!settingsResult.success) {
|
||||||
|
return reply.status(400).send({
|
||||||
|
error: `Step ${i + 1} (${step.toolId}): Invalid settings`,
|
||||||
|
details: settingsResult.error.issues.map((iss: { path: (string | number)[]; message: string }) => ({
|
||||||
|
path: iss.path.join("."),
|
||||||
|
message: iss.message,
|
||||||
|
})),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Execute the pipeline: pass the buffer through each step sequentially
|
||||||
|
let currentBuffer = fileBuffer;
|
||||||
|
let currentFilename = filename;
|
||||||
|
const stepResults: Array<{ step: number; toolId: string; size: number }> = [];
|
||||||
|
|
||||||
|
try {
|
||||||
|
for (let i = 0; i < pipeline.steps.length; i++) {
|
||||||
|
const step = pipeline.steps[i];
|
||||||
|
const toolConfig = getToolConfig(step.toolId)!;
|
||||||
|
|
||||||
|
// Parse settings through the schema to apply defaults
|
||||||
|
const settings = toolConfig.settingsSchema.parse(step.settings);
|
||||||
|
|
||||||
|
const result = await toolConfig.process(currentBuffer, settings, currentFilename);
|
||||||
|
|
||||||
|
stepResults.push({
|
||||||
|
step: i + 1,
|
||||||
|
toolId: step.toolId,
|
||||||
|
size: result.buffer.length,
|
||||||
|
});
|
||||||
|
|
||||||
|
currentBuffer = result.buffer;
|
||||||
|
currentFilename = result.filename;
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
const message = err instanceof Error ? err.message : "Pipeline processing failed";
|
||||||
|
return reply.status(422).send({
|
||||||
|
error: "Pipeline processing failed",
|
||||||
|
details: message,
|
||||||
|
completedSteps: stepResults,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
// Save the final output to workspace
|
||||||
|
const jobId = randomUUID();
|
||||||
|
const workspacePath = await createWorkspace(jobId);
|
||||||
|
const outputPath = join(workspacePath, "output", currentFilename);
|
||||||
|
await writeFile(outputPath, currentBuffer);
|
||||||
|
|
||||||
|
// Also save the original input for reference
|
||||||
|
const inputPath = join(workspacePath, "input", filename);
|
||||||
|
await writeFile(inputPath, fileBuffer);
|
||||||
|
|
||||||
|
return reply.send({
|
||||||
|
jobId,
|
||||||
|
downloadUrl: `/api/v1/download/${jobId}/${encodeURIComponent(currentFilename)}`,
|
||||||
|
originalSize: fileBuffer.length,
|
||||||
|
processedSize: currentBuffer.length,
|
||||||
|
stepsCompleted: stepResults.length,
|
||||||
|
steps: stepResults,
|
||||||
|
});
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* POST /api/v1/pipeline/save
|
||||||
|
*
|
||||||
|
* Save a named pipeline definition for later reuse.
|
||||||
|
*/
|
||||||
|
app.post(
|
||||||
|
"/api/v1/pipeline/save",
|
||||||
|
async (request: FastifyRequest, reply: FastifyReply) => {
|
||||||
|
const body = request.body as unknown;
|
||||||
|
const result = savePipelineSchema.safeParse(body);
|
||||||
|
|
||||||
|
if (!result.success) {
|
||||||
|
return reply.status(400).send({
|
||||||
|
error: "Invalid pipeline definition",
|
||||||
|
details: result.error.issues.map((i) => ({
|
||||||
|
path: i.path.join("."),
|
||||||
|
message: i.message,
|
||||||
|
})),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
const { name, description, steps } = result.data;
|
||||||
|
|
||||||
|
// Validate all tool IDs exist
|
||||||
|
for (let i = 0; i < steps.length; i++) {
|
||||||
|
const toolConfig = getToolConfig(steps[i].toolId);
|
||||||
|
if (!toolConfig) {
|
||||||
|
return reply.status(400).send({
|
||||||
|
error: `Step ${i + 1}: Tool "${steps[i].toolId}" not found`,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const id = randomUUID();
|
||||||
|
|
||||||
|
db.insert(schema.pipelines)
|
||||||
|
.values({
|
||||||
|
id,
|
||||||
|
name,
|
||||||
|
description: description ?? null,
|
||||||
|
steps: JSON.stringify(steps),
|
||||||
|
})
|
||||||
|
.run();
|
||||||
|
|
||||||
|
return reply.status(201).send({
|
||||||
|
id,
|
||||||
|
name,
|
||||||
|
description: description ?? null,
|
||||||
|
steps,
|
||||||
|
createdAt: new Date().toISOString(),
|
||||||
|
});
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* GET /api/v1/pipeline/list
|
||||||
|
*
|
||||||
|
* List all saved pipelines.
|
||||||
|
*/
|
||||||
|
app.get(
|
||||||
|
"/api/v1/pipeline/list",
|
||||||
|
async (_request: FastifyRequest, reply: FastifyReply) => {
|
||||||
|
const rows = db.select().from(schema.pipelines).all();
|
||||||
|
|
||||||
|
const pipelines = rows.map((row) => ({
|
||||||
|
id: row.id,
|
||||||
|
name: row.name,
|
||||||
|
description: row.description,
|
||||||
|
steps: JSON.parse(row.steps),
|
||||||
|
createdAt: row.createdAt.toISOString(),
|
||||||
|
}));
|
||||||
|
|
||||||
|
return reply.send({ pipelines });
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* DELETE /api/v1/pipeline/:id
|
||||||
|
*
|
||||||
|
* Delete a saved pipeline by its ID.
|
||||||
|
*/
|
||||||
|
app.delete(
|
||||||
|
"/api/v1/pipeline/:id",
|
||||||
|
async (
|
||||||
|
request: FastifyRequest<{ Params: { id: string } }>,
|
||||||
|
reply: FastifyReply,
|
||||||
|
) => {
|
||||||
|
const { id } = request.params;
|
||||||
|
|
||||||
|
const existing = db
|
||||||
|
.select()
|
||||||
|
.from(schema.pipelines)
|
||||||
|
.where(eq(schema.pipelines.id, id))
|
||||||
|
.get();
|
||||||
|
|
||||||
|
if (!existing) {
|
||||||
|
return reply.status(404).send({ error: "Pipeline not found" });
|
||||||
|
}
|
||||||
|
|
||||||
|
db.delete(schema.pipelines)
|
||||||
|
.where(eq(schema.pipelines.id, id))
|
||||||
|
.run();
|
||||||
|
|
||||||
|
return reply.send({ ok: true });
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
app.log.info("Pipeline routes registered");
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user