External/SnapOtter
1
0
Fork 0
mirror of https://github.com/snapotter-hq/SnapOtter.git synced 2026-08-03 07:46:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(analytics): build-time bake + telemetry depth (#336)

Browse Source 
Bake PostHog + Sentry into the published Docker image (SNAPOTTER_ANALYTICS
build arg, codegen script). Delete entire consent system. Move event emission
to BullMQ worker. Add cross-tier identity stitching, Sentry performance
tracing on both tiers, frontend funnel events. Fix stateful regex bug.

86 files changed, 1593 insertions(+), 3747 deletions(-)
This commit is contained in:
SnapOtter
2026-06-24 11:05:39 +08:00
committed by GitHub
parent a53038ed96
commit 5d36ac06d8
86 changed files with 1594 additions and 3748 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/db/schema.ts
-3
View File
@@ -41,9 +41,6 @@ export const users = pgTable("users", {
totpSecret: text("totp_secret"),
totpEnabled: boolean("totp_enabled").notNull().default(false),
recoveryCodesHash: text("recovery_codes_hash"),
analyticsEnabled: boolean("analytics_enabled"),
analyticsConsentShownAt: timestamp("analytics_consent_shown_at", { withTimezone: true }),
analyticsConsentRemindAt: timestamp("analytics_consent_remind_at", { withTimezone: true }),
});
export const teams = pgTable("teams", {
apps/api/src/index.ts
+1 -1
View File
@@ -254,7 +254,7 @@ app.setErrorHandler((error: Error & { statusCode?: number }, request, reply) =>
{ err: error, url: request.url, method: request.method },
"Unhandled request error",
);
captureException(error, request);
captureException(error);
} else {
request.log.warn({ err: error, url: request.url, method: request.method }, "Request error");
}
apps/api/src/instrument.ts
+67
View File
@@ -0,0 +1,67 @@
import { ANALYTICS_BAKED } from "@snapotter/shared";
const FILE_EXT_PATTERN =
/\.(jpe?g|png|pdf|webp|gif|tiff?|bmp|svg|hei[cf]?|avif|raw|cr2|nef|arw|dng|psd|tga|exr|hdr)\b/gi;
const FILE_PATH_PATTERN = /\/(tmp\/workspace|data\/files|data\/ai)\//g;
if (ANALYTICS_BAKED.enabled && ANALYTICS_BAKED.sentryDsn) {
try {
const Sentry = await import("@sentry/node");
const { APP_VERSION } = await import("@snapotter/shared");
Sentry.init({
dsn: ANALYTICS_BAKED.sentryDsn,
release: APP_VERSION,
environment: process.env.NODE_ENV || "production",
tracesSampleRate: ANALYTICS_BAKED.sampleRate,
sendDefaultPii: false,
beforeSend(event) {
if (event.user) {
delete event.user.email;
delete event.user.username;
}
if (event.exception?.values) {
for (const ex of event.exception.values) {
if (
ex.value &&
(ex.value.includes("Rate limit exceeded") ||
ex.value.includes("Body cannot be empty") ||
ex.value.includes("Unsupported Media Type") ||
ex.value.includes("Request body size did not match") ||
ex.value.includes("Premature close"))
) {
return null;
}
if (ex.value) {
ex.value = ex.value
.replace(FILE_EXT_PATTERN, ".[REDACTED]")
.replace(FILE_PATH_PATTERN, "/[REDACTED]/");
}
if (ex.stacktrace?.frames) {
for (const frame of ex.stacktrace.frames) {
if (frame.filename) {
frame.filename = frame.filename
.replace(FILE_EXT_PATTERN, ".[REDACTED]")
.replace(FILE_PATH_PATTERN, "/[REDACTED]/");
}
}
}
}
}
return event;
},
beforeBreadcrumb(breadcrumb) {
if (breadcrumb.message) {
breadcrumb.message = breadcrumb.message
.replace(FILE_EXT_PATTERN, ".[REDACTED]")
.replace(FILE_PATH_PATTERN, "/[REDACTED]/");
}
return breadcrumb;
},
});
console.log("[sentry] initialized with performance tracing, release:", APP_VERSION);
} catch {
// @sentry/node not available
}
}
apps/api/src/jobs/types.ts
+1
View File
@@ -46,6 +46,7 @@ export interface ToolJobData {
parentId?: string;
totalFiles?: number;
fileIndex?: number;
analyticsDistinctId?: string;
_otel?: { traceparent: string; tracestate?: string };
}
apps/api/src/jobs/worker.ts
+81
View File
@@ -25,10 +25,12 @@ import { mkdir, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { context, propagation, ROOT_CONTEXT, SpanStatusCode, trace } from "@opentelemetry/api";
import { ANALYTICS_BAKED, ANALYTICS_EVENTS, getBundleForTool, TOOLS } from "@snapotter/shared";
import { type Job, UnrecoverableError, Worker } from "bullmq";
import { eq } from "drizzle-orm";
import { env } from "../config.js";
import { db, schema } from "../db/index.js";
import { captureException, trackEvent } from "../lib/analytics.js";
import { resolveConcurrency } from "../lib/env.js";
import { friendlyError } from "../lib/errors.js";
import { logger } from "../lib/logger.js";
@@ -312,6 +314,22 @@ async function processToolJob(job: Job<ToolJobData>): Promise<ToolJobResult> {
jobsTotal.inc({ pool: data.pool, status: "completed" });
jobDuration.observe({ pool: data.pool }, durationMs / 1000);
// Analytics: emit tool_used on success
if (ANALYTICS_BAKED.enabled) {
const tool = TOOLS.find((t) => t.id === data.toolId);
void trackEvent(
ANALYTICS_EVENTS.TOOL_USED,
{
tool_id: data.toolId,
status: "completed",
duration_ms: durationMs,
category: tool?.category ?? "unknown",
is_ai_tool: getBundleForTool(data.toolId) !== null,
},
data.analyticsDistinctId,
);
}
// Emit terminal progress event with legacy result payload
const legacyResult = buildLegacyResultPayload(jobResult, jobId);
updateSingleFileProgress({
@@ -401,6 +419,26 @@ async function processToolJob(job: Job<ToolJobData>): Promise<ToolJobResult> {
}
}
// Analytics: emit tool_used on failure
if (ANALYTICS_BAKED.enabled) {
const tool = TOOLS.find((t) => t.id === data.toolId);
void trackEvent(
ANALYTICS_EVENTS.TOOL_USED,
{
tool_id: data.toolId,
status: "failed",
duration_ms: durationMs,
category: tool?.category ?? "unknown",
is_ai_tool: getBundleForTool(data.toolId) !== null,
error_code: isTimeout ? "timeout" : isCanceled ? "cancelled" : "processing",
},
data.analyticsDistinctId,
);
if (!isCanceled && !isTimeout) {
void captureException(err instanceof Error ? err : new Error(String(err)));
}
}
if (isCanceled) throw new UnrecoverableError("Canceled");
if (isTimeout) throw new Error(finalError);
throw err;
@@ -547,6 +585,7 @@ function contentTypeForFilename(name: string): string {
async function processPipelineFinalize(job: Job<ToolJobData>): Promise<ToolJobResult> {
const data = job.data;
const startTime = Date.now();
const totalSteps = data.totalSteps ?? 0;
const steps: Array<{ step: number; toolId: string; size: number }> = [];
@@ -608,6 +647,21 @@ async function processPipelineFinalize(job: Job<ToolJobData>): Promise<ToolJobRe
await recordChildOutcome(data.parentId, data.totalFiles, data.filename, errorMsg);
}
// Analytics: emit pipeline_executed on failure
if (ANALYTICS_BAKED.enabled) {
void trackEvent(
ANALYTICS_EVENTS.PIPELINE_EXECUTED,
{
step_count: totalSteps,
tool_ids: steps.map((s) => s.toolId),
is_batch: data.kind === "batch-finalize",
duration_ms: Date.now() - startTime,
status: "failed",
},
data.analyticsDistinctId,
);
}
return {
outputRefs: [],
filename: data.filename,
@@ -677,6 +731,21 @@ async function processPipelineFinalize(job: Job<ToolJobData>): Promise<ToolJobRe
await recordChildOutcome(data.parentId, data.totalFiles, outFilename);
}
// Analytics: emit pipeline_executed on success
if (ANALYTICS_BAKED.enabled) {
void trackEvent(
ANALYTICS_EVENTS.PIPELINE_EXECUTED,
{
step_count: totalSteps,
tool_ids: steps.map((s) => s.toolId),
is_batch: data.kind === "batch-finalize",
duration_ms: Date.now() - startTime,
status: "completed",
},
data.analyticsDistinctId,
);
}
return result;
}
@@ -805,6 +874,12 @@ export function startWorkers(): void {
logger.error({ err, pool }, "Worker error");
});
worker.on("failed", (job, err) => {
if (ANALYTICS_BAKED.enabled && job) {
void captureException(err instanceof Error ? err : new Error(String(err)));
}
});
workers.push(worker);
continue;
}
@@ -827,6 +902,12 @@ export function startWorkers(): void {
logger.error({ err, pool }, "Worker error");
});
worker.on("failed", (job, err) => {
if (ANALYTICS_BAKED.enabled && job) {
void captureException(err instanceof Error ? err : new Error(String(err)));
}
});
workers.push(worker);
}
apps/api/src/lib/analytics.ts
+19 -105
View File
@@ -1,92 +1,32 @@
import { ANALYTICS_BAKED } from "@snapotter/shared";
import { eq } from "drizzle-orm";
import type { FastifyRequest } from "fastify";
import type { PostHog } from "posthog-node";
import { env } from "../config.js";
import { db, schema } from "../db/index.js";
import { getAuthUser } from "../plugins/auth.js";
const FILE_EXT_PATTERN =
/\.(jpe?g|png|pdf|webp|gif|tiff?|bmp|svg|hei[cf]?|avif|raw|cr2|nef|arw|dng|psd|tga|exr|hdr)\b/gi;
const FILE_PATH_PATTERN = /\/(tmp\/workspace|data\/files|data\/ai)\//g;
let posthogClient: PostHog | null = null;
let sentryModule: typeof import("@sentry/node") | null = null;
export async function initAnalytics(): Promise<void> {
if (!env.ANALYTICS_ENABLED || !env.POSTHOG_API_KEY) return;
if (!ANALYTICS_BAKED.enabled) return;
try {
const { PostHog } = await import("posthog-node");
posthogClient = new PostHog(env.POSTHOG_API_KEY, {
host: env.POSTHOG_HOST,
flushAt: 20,
flushInterval: 30000,
});
} catch {
// posthog-node not available — analytics disabled
}
if (env.SENTRY_DSN) {
if (ANALYTICS_BAKED.posthogApiKey) {
try {
sentryModule = await import("@sentry/node");
sentryModule.init({
dsn: env.SENTRY_DSN,
sendDefaultPii: false,
beforeSend(event) {
if (event.user) {
delete event.user.email;
delete event.user.username;
}
if (event.exception?.values) {
for (const ex of event.exception.values) {
if (
ex.value &&
(ex.value.includes("Rate limit exceeded") ||
ex.value.includes("Body cannot be empty") ||
ex.value.includes("Unsupported Media Type") ||
ex.value.includes("Request body size did not match") ||
ex.value.includes("Premature close"))
) {
return null;
}
if (ex.value) {
ex.value = ex.value
.replace(FILE_EXT_PATTERN, ".[REDACTED]")
.replace(FILE_PATH_PATTERN, "/[REDACTED]/");
}
if (ex.stacktrace?.frames) {
for (const frame of ex.stacktrace.frames) {
if (frame.filename) {
frame.filename = frame.filename
.replace(FILE_EXT_PATTERN, ".[REDACTED]")
.replace(FILE_PATH_PATTERN, "/[REDACTED]/");
}
}
}
}
}
return event;
},
beforeBreadcrumb(breadcrumb) {
if (breadcrumb.message) {
breadcrumb.message = breadcrumb.message
.replace(FILE_EXT_PATTERN, ".[REDACTED]")
.replace(FILE_PATH_PATTERN, "/[REDACTED]/");
}
return breadcrumb;
},
const { PostHog } = await import("posthog-node");
posthogClient = new PostHog(ANALYTICS_BAKED.posthogApiKey, {
host: ANALYTICS_BAKED.posthogHost,
flushAt: 20,
flushInterval: 30000,
});
} catch {
// @sentry/node not available
// posthog-node not available
}
}
}
export async function captureException(error: unknown, request?: FastifyRequest): Promise<void> {
export async function captureException(error: unknown): Promise<void> {
try {
if (!sentryModule) return;
if (request && !(await isRequestOptedIn(request))) return;
sentryModule.captureException(error);
if (!ANALYTICS_BAKED.enabled) return;
const Sentry = await import("@sentry/node");
Sentry.captureException(error);
} catch {
// analytics must never throw
}
@@ -97,11 +37,6 @@ export async function shutdownAnalytics(): Promise<void> {
await posthogClient.shutdown();
posthogClient = null;
}
if (sentryModule) {
await sentryModule.close(2000);
sentryModule = null;
}
}
async function getInstanceId(): Promise<string> {
@@ -112,39 +47,18 @@ async function getInstanceId(): Promise<string> {
return row?.value ?? "unknown";
}
async function isUserOptedIn(userId: string): Promise<boolean> {
if (!env.ANALYTICS_ENABLED) return false;
if (userId === "anonymous") return false;
const [user] = await db.select().from(schema.users).where(eq(schema.users.id, userId));
return user?.analyticsEnabled === true;
}
async function isRequestOptedIn(request: FastifyRequest): Promise<boolean> {
if (!env.ANALYTICS_ENABLED) return false;
const user = getAuthUser(request);
if (!user) return false;
if (user.id === "anonymous") {
const header = request.headers["x-analytics-consent"];
return header === "true";
}
return await isUserOptedIn(user.id);
}
function shouldSample(): boolean {
if (env.ANALYTICS_SAMPLE_RATE >= 1.0) return true;
if (env.ANALYTICS_SAMPLE_RATE <= 0.0) return false;
return Math.random() < env.ANALYTICS_SAMPLE_RATE;
}
export async function trackEvent(
request: FastifyRequest,
event: string,
properties: Record<string, unknown>,
distinctId?: string,
): Promise<void> {
try {
if (!posthogClient || !(await isRequestOptedIn(request)) || !shouldSample()) return;
if (!ANALYTICS_BAKED.enabled || !posthogClient) return;
if (ANALYTICS_BAKED.sampleRate < 1.0) {
if (ANALYTICS_BAKED.sampleRate <= 0.0 || Math.random() >= ANALYTICS_BAKED.sampleRate) return;
}
posthogClient.capture({
distinctId: await getInstanceId(),
distinctId: distinctId ?? (await getInstanceId()),
event,
properties,
});
apps/api/src/lib/env.ts
-8
View File
@@ -110,14 +110,6 @@ const envSchema = z
AUDIT_RETENTION_DAYS: z.coerce.number().default(0),
LOG_DIR: z.string().default("./data/logs"),
SCRATCH_PATH: z.string().default(""),
ANALYTICS_ENABLED: z
.enum(["true", "false"])
.default("false")
.transform((v) => v === "true"),
ANALYTICS_SAMPLE_RATE: z.coerce.number().min(0).max(1).default(1.0),
POSTHOG_API_KEY: z.string().default(""),
POSTHOG_HOST: z.string().default("https://us.i.posthog.com"),
SENTRY_DSN: z.string().default(""),
DATA_ENCRYPTION_KEY: z.string().default(""),
DATA_ENCRYPTION_KEY_PREVIOUS: z.string().default(""),
})
apps/api/src/plugins/auth.ts
-9
View File
@@ -435,9 +435,6 @@ export async function authRoutes(app: FastifyInstance): Promise<void> {
mustChangePassword: env.SKIP_MUST_CHANGE_PASSWORD ? false : user.mustChangePassword,
permissions: await getPermissions(user.role),
teamName: teamRow?.name ?? user.team,
analyticsEnabled: user.analyticsEnabled ?? null,
analyticsConsentShownAt: user.analyticsConsentShownAt?.getTime() ?? null,
analyticsConsentRemindAt: user.analyticsConsentRemindAt?.getTime() ?? null,
},
expiresAt: expiresAt.toISOString(),
...(mfaRequired && { mfaRequired: true }),
@@ -498,9 +495,6 @@ export async function authRoutes(app: FastifyInstance): Promise<void> {
role: "admin",
mustChangePassword: false,
permissions: await getPermissions("admin"),
analyticsEnabled: null,
analyticsConsentShownAt: null,
analyticsConsentRemindAt: null,
},
expiresAt: null,
});
@@ -538,9 +532,6 @@ export async function authRoutes(app: FastifyInstance): Promise<void> {
email: user.email ?? null,
hasLocalPassword: !!user.passwordHash,
hasOidcLink: !!user.externalId,
analyticsEnabled: user.analyticsEnabled ?? null,
analyticsConsentShownAt: user.analyticsConsentShownAt?.getTime() ?? null,
analyticsConsentRemindAt: user.analyticsConsentRemindAt?.getTime() ?? null,
},
expiresAt: session.expiresAt.toISOString(),
});
apps/api/src/plugins/mfa.ts
-3
View File
@@ -350,9 +350,6 @@ export async function registerMfa(app: FastifyInstance): Promise<void> {
mustChangePassword: env.SKIP_MUST_CHANGE_PASSWORD ? false : dbUser.mustChangePassword,
permissions: await getPermissions(dbUser.role),
teamName: teamRow?.name ?? dbUser.team,
analyticsEnabled: dbUser.analyticsEnabled ?? null,
analyticsConsentShownAt: dbUser.analyticsConsentShownAt?.getTime() ?? null,
analyticsConsentRemindAt: dbUser.analyticsConsentRemindAt?.getTime() ?? null,
},
expiresAt: expiresAt.toISOString(),
});
apps/api/src/routes/analytics.ts
+6 -70
View File
@@ -1,86 +1,22 @@
import { ANALYTICS_BAKED } from "@snapotter/shared";
import { eq } from "drizzle-orm";
import type { FastifyInstance } from "fastify";
import { z } from "zod";
import { env } from "../config.js";
import { db, schema } from "../db/index.js";
import { requireAuth } from "../plugins/auth.js";
const analyticsConsentSchema = z.object({
enabled: z.boolean().optional(),
remindLater: z.boolean().optional(),
});
export async function analyticsRoutes(app: FastifyInstance): Promise<void> {
app.get("/api/v1/config/analytics", async () => {
if (!env.ANALYTICS_ENABLED) {
return {
enabled: false,
posthogApiKey: "",
posthogHost: "",
sentryDsn: "",
sampleRate: 0,
instanceId: "",
};
}
const [row] = await db
.select()
.from(schema.settings)
.where(eq(schema.settings.key, "instance_id"));
return {
enabled: true,
posthogApiKey: env.POSTHOG_API_KEY,
posthogHost: env.POSTHOG_HOST,
sentryDsn: env.SENTRY_DSN,
sampleRate: env.ANALYTICS_SAMPLE_RATE,
enabled: ANALYTICS_BAKED.enabled,
posthogApiKey: ANALYTICS_BAKED.posthogApiKey,
posthogHost: ANALYTICS_BAKED.posthogHost,
sentryDsn: ANALYTICS_BAKED.sentryDsn,
sampleRate: ANALYTICS_BAKED.sampleRate,
instanceId: row?.value ?? "",
};
});
app.put(
"/api/v1/user/analytics",
{ config: { rateLimit: { max: 300, timeWindow: "1 minute" } } },
async (request, reply) => {
const user = requireAuth(request, reply);
if (!user) return;
const parsed = analyticsConsentSchema.safeParse(request.body ?? {});
if (!parsed.success) {
return reply.status(400).send({
error: parsed.error.issues.map((i) => i.message).join("; "),
code: "VALIDATION_ERROR",
});
}
const body = parsed.data;
const now = new Date();
if (body.remindLater) {
const remindAt = new Date(Date.now() + 7 * 24 * 60 * 60 * 1000);
await db
.update(schema.users)
.set({
analyticsEnabled: null,
analyticsConsentShownAt: now,
analyticsConsentRemindAt: remindAt,
updatedAt: now,
})
.where(eq(schema.users.id, user.id));
return reply.send({ ok: true, analyticsEnabled: null });
}
const enabled = body.enabled === true;
await db
.update(schema.users)
.set({
analyticsEnabled: enabled,
analyticsConsentShownAt: now,
analyticsConsentRemindAt: null,
updatedAt: now,
})
.where(eq(schema.users.id, user.id));
return reply.send({ ok: true, analyticsEnabled: enabled });
},
);
}
apps/api/src/routes/batch.ts
+2
View File
@@ -304,6 +304,7 @@ export async function registerBatchRoutes(app: FastifyInstance): Promise<void> {
inputRefs: [key],
filename: processFilename,
settings,
analyticsDistinctId: request.headers["x-posthog-distinct-id"] as string | undefined,
} satisfies ToolJobData,
// Children swallow failures via return markers, so a retry would
// never run; attempts: 1 makes that explicit.
@@ -340,6 +341,7 @@ export async function registerBatchRoutes(app: FastifyInstance): Promise<void> {
inputRefs: [],
filename: "",
settings: { flowChildCount: flowChildren.length },
analyticsDistinctId: request.headers["x-posthog-distinct-id"] as string | undefined,
} satisfies ToolJobData,
opts: { jobId: parentId, attempts: 1 },
children: flowChildren,
apps/api/src/routes/features.ts
+3 -4
View File
@@ -165,7 +165,6 @@ export async function registerFeatureRoutes(app: FastifyInstance): Promise<void>
const modelsDir = getModelsDir();
const installStartTime = Date.now();
const reqRef = request;
// Hold the venv lock across the whole install so no AI tool job loads
// native libs from the venv while pip is rewriting them (that segfaults
@@ -240,7 +239,7 @@ export async function registerFeatureRoutes(app: FastifyInstance): Promise<void>
shutdownDispatcher();
setInstallProgress(null, null, null);
updateSingleFileProgress({ jobId, phase: "complete", percent: 100, stage: "Complete" });
trackEvent(reqRef, ANALYTICS_EVENTS.AI_BUNDLE_ACTION, {
trackEvent(ANALYTICS_EVENTS.AI_BUNDLE_ACTION, {
bundle_id: bundleId,
action: "installed",
duration_ms: Date.now() - installStartTime,
@@ -366,7 +365,7 @@ export async function registerFeatureRoutes(app: FastifyInstance): Promise<void>
markUninstalled(bundleId);
shutdownDispatcher();
trackEvent(request, ANALYTICS_EVENTS.AI_BUNDLE_ACTION, {
trackEvent(ANALYTICS_EVENTS.AI_BUNDLE_ACTION, {
bundle_id: bundleId,
action: "uninstalled",
duration_ms: 0,
@@ -410,7 +409,7 @@ export async function registerFeatureRoutes(app: FastifyInstance): Promise<void>
const result = await importBundleArchive(part.file);
invalidateCache();
shutdownDispatcher();
trackEvent(request, ANALYTICS_EVENTS.AI_BUNDLE_ACTION, {
trackEvent(ANALYTICS_EVENTS.AI_BUNDLE_ACTION, {
bundle_id: result.bundleId,
action: "imported",
duration_ms: 0,
apps/api/src/routes/pipeline.ts
+9 -66
View File
@@ -11,7 +11,7 @@ import { randomUUID } from "node:crypto";
import { mkdir } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { ANALYTICS_EVENTS, FEATURE_BUNDLES, MODALITY_POOL, TOOLS } from "@snapotter/shared";
import { FEATURE_BUNDLES, MODALITY_POOL, TOOLS } from "@snapotter/shared";
import archiver from "archiver";
import type { FlowJob } from "bullmq";
import { eq } from "drizzle-orm";
@@ -22,7 +22,6 @@ import { db, schema } from "../db/index.js";
import { recordChildOutcome } from "../jobs/batch-progress.js";
import { getFlowProducer, injectTraceContext, waitForJob } from "../jobs/enqueue.js";
import { type Pool, queueName, type ToolJobData } from "../jobs/types.js";
import { trackEvent } from "../lib/analytics.js";
import { autoOrient } from "../lib/auto-orient.js";
import { getSecurityHeaders } from "../lib/csp.js";
import { formatZodErrors } from "../lib/errors.js";
@@ -115,6 +114,7 @@ function buildPipelineFlowTree(opts: {
clientJobId?: string;
parentId?: string;
totalFiles?: number;
analyticsDistinctId?: string;
}): { tree: FlowJob; stepJobIds: string[] } {
const {
jobId,
@@ -126,6 +126,7 @@ function buildPipelineFlowTree(opts: {
clientJobId,
parentId,
totalFiles,
analyticsDistinctId,
} = opts;
const totalSteps = parsedSteps.length;
const stepJobIds = parsedSteps.map((_: unknown, i: number) => `${jobId}-s${i}`);
@@ -149,6 +150,7 @@ function buildPipelineFlowTree(opts: {
inputRefs: [uploadKey],
filename,
settings: parsedSteps[0].parsedSettings,
analyticsDistinctId,
} satisfies ToolJobData,
opts: { jobId: stepJobIds[0], attempts: 1 },
};
@@ -170,6 +172,7 @@ function buildPipelineFlowTree(opts: {
inputRefs: [],
filename,
settings: parsedSteps[i].parsedSettings,
analyticsDistinctId,
} satisfies ToolJobData,
opts: { jobId: stepJobIds[i], attempts: 1 },
children: [currentNode],
@@ -195,6 +198,7 @@ function buildPipelineFlowTree(opts: {
inputRefs: [],
filename,
settings: {},
analyticsDistinctId,
} satisfies ToolJobData,
opts: { jobId, attempts: 1 },
children: [currentNode],
@@ -423,7 +427,6 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
// ── Enqueue as a BullMQ flow ────────────────────────────────
const startTime = Date.now();
const jobId = randomUUID();
const userId = getAuthUser(request)?.id ?? null;
const originalSize = fileBuffer.length;
@@ -457,6 +460,7 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
filename,
pipelinePool,
clientJobId: clientJobId ?? jobId,
analyticsDistinctId: request.headers["x-posthog-distinct-id"] as string | undefined,
});
// Insert all durable rows before adding the flow. enqueueToolJob
@@ -497,13 +501,6 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
const result = await waitForJob(pipelinePool, jobId, 10 * 60_000);
if (!result) {
trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, {
step_count: pipeline.steps.length,
tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId),
is_batch: false,
duration_ms: Date.now() - startTime,
status: "failed",
});
return reply.status(422).send({
error: "Pipeline processing timed out",
});
@@ -511,27 +508,12 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
// Check for step failure reported by the finalize handler
if (result.resultPayload?.error) {
trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, {
step_count: pipeline.steps.length,
tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId),
is_batch: false,
duration_ms: Date.now() - startTime,
status: "failed",
});
return reply.status(422).send({
error: result.resultPayload.error as string,
completedSteps: result.resultPayload.steps,
});
}
trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, {
step_count: pipeline.steps.length,
tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId),
is_batch: false,
duration_ms: Date.now() - startTime,
status: "completed",
});
return reply.send({
jobId,
downloadUrl: `/api/v1/download/${jobId}/${encodeURIComponent(result.filename)}`,
@@ -544,13 +526,6 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
steps: result.resultPayload?.steps ?? [],
});
} catch (err) {
trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, {
step_count: pipeline.steps.length,
tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId),
is_batch: false,
duration_ms: Date.now() - startTime,
status: "failed",
});
return reply.status(422).send({
error: err instanceof Error ? err.message : "Pipeline processing failed",
});
@@ -856,7 +831,6 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
}
// ── Prepare files and build flow ─────────────────────────────────
const batchStartTime = Date.now();
const parentId = clientJobId || randomUUID();
const userId = getAuthUser(request)?.id ?? null;
@@ -990,6 +964,7 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
pipelinePool: batchPipelinePool,
parentId,
totalFiles: files.length,
analyticsDistinctId: request.headers["x-posthog-distinct-id"] as string | undefined,
});
// Insert step + finalize rows for this file
@@ -1028,14 +1003,6 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
if (perFileChildren.length === 0) {
// All files failed validation
trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, {
step_count: pipeline.steps.length,
tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId),
is_batch: true,
file_count: files.length,
duration_ms: Date.now() - batchStartTime,
status: "failed",
});
return reply.status(422).send({
error: "All files failed processing",
errors: preFailures.map((f) => ({ filename: f.filename, error: f.error })),
@@ -1056,6 +1023,7 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
inputRefs: [],
filename: "",
settings: { flowChildCount: perFileChildren.length },
analyticsDistinctId: request.headers["x-posthog-distinct-id"] as string | undefined,
} satisfies ToolJobData,
opts: { jobId: parentId, attempts: 1 },
children: perFileChildren,
@@ -1076,14 +1044,6 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
const batchResult = await waitForJob("system", parentId, 30 * 60_000);
if (!batchResult) {
trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, {
step_count: pipeline.steps.length,
tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId),
is_batch: true,
file_count: files.length,
duration_ms: Date.now() - batchStartTime,
status: "failed",
});
return reply.status(422).send({ error: "Pipeline batch processing timed out" });
}
@@ -1150,29 +1110,12 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise<void
const failedEntries = allResults.filter((r) => !r.outputRef);
if (successEntries.length === 0) {
trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, {
step_count: pipeline.steps.length,
tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId),
is_batch: true,
file_count: files.length,
duration_ms: Date.now() - batchStartTime,
status: "failed",
});
return reply.status(422).send({
error: "All files failed processing",
errors: failedEntries.map((f) => ({ filename: f.filename, error: f.error ?? "Failed" })),
});
}
trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, {
step_count: pipeline.steps.length,
tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId),
is_batch: true,
file_count: files.length,
duration_ms: Date.now() - batchStartTime,
status: "completed",
});
const fileResultsMap: Record<string, string> = {};
for (const entry of successEntries) {
const uniqueName = getUniqueName(entry.filename);
apps/api/src/routes/tool-factory.ts
+1 -19
View File
@@ -3,7 +3,6 @@ import { mkdir, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { extname, join } from "node:path";
import {
ANALYTICS_EVENTS,
apiToolPath,
getBundleForTool,
type Section,
@@ -16,7 +15,6 @@ import type { z } from "zod";
import { env } from "../config.js";
import { db, schema } from "../db/index.js";
import { enqueueToolJob, waitForJob } from "../jobs/enqueue.js";
import { trackEvent } from "../lib/analytics.js";
import { formatZodErrors, friendlyError, stripInternalPaths } from "../lib/errors.js";
import { isToolInstalled } from "../lib/feature-status.js";
import { getObjectBuffer, putObject } from "../lib/object-storage.js";
@@ -556,6 +554,7 @@ export function createToolRoute<T>(app: FastifyInstance, config: ToolRouteConfig
fileId: fileId ?? undefined,
clientJobId: clientJobId ?? undefined,
kind: "tool",
analyticsDistinctId: request.headers["x-posthog-distinct-id"] as string | undefined,
});
// Long tools never block the HTTP request (spec 4.5): straight to SSE.
@@ -566,14 +565,6 @@ export function createToolRoute<T>(app: FastifyInstance, config: ToolRouteConfig
try {
const result = await waitForJob(pool, jobId);
if (result) {
trackEvent(request, ANALYTICS_EVENTS.TOOL_USED, {
tool_id: config.toolId,
status: "completed",
duration_ms: Date.now() - startTime,
category: TOOLS.find((t) => t.id === config.toolId)?.category ?? "unknown",
is_ai_tool: getBundleForTool(config.toolId) !== null,
});
// Fire-and-forget: audit log must never block the response
import("../lib/audit.js")
.then(({ isToolAuditEnabled, auditFromRequest }) =>
@@ -608,15 +599,6 @@ export function createToolRoute<T>(app: FastifyInstance, config: ToolRouteConfig
}
return reply.status(202).send({ jobId: clientJobId || jobId, async: true });
} catch (err) {
trackEvent(request, ANALYTICS_EVENTS.TOOL_USED, {
tool_id: config.toolId,
status: "failed",
duration_ms: Date.now() - startTime,
category: TOOLS.find((t) => t.id === config.toolId)?.category ?? "unknown",
is_ai_tool: getBundleForTool(config.toolId) !== null,
error_code: err instanceof Error ? err.constructor.name : "UnknownError",
error_message: err instanceof Error ? err.message.slice(0, 200) : "Processing failed",
});
// Keep the full error (incl. raw ffmpeg/tool stderr) in server logs,
// but return only a user-safe detail to the client.
request.log.error({ err, toolId: config.toolId }, "tool processing failed");