Instanzweite Scan-Agenten vorbereiten
This commit is contained in:
235
backend/src/routes/instanceAgentGateway.ts
Normal file
235
backend/src/routes/instanceAgentGateway.ts
Normal file
@@ -0,0 +1,235 @@
|
||||
import { FastifyInstance, FastifyRequest } from "fastify"
|
||||
import multipart from "@fastify/multipart"
|
||||
import { createHash } from "node:crypto"
|
||||
import { and, asc, eq, sql } from "drizzle-orm"
|
||||
import { instanceAgentScanJobs, instanceAgents } from "../../db/schema"
|
||||
import { saveFile } from "../utils/files"
|
||||
|
||||
const hashToken = (token: string) =>
|
||||
createHash("sha256").update(token, "utf8").digest("hex")
|
||||
|
||||
const readAgentToken = (req: FastifyRequest) => {
|
||||
const headerToken = req.headers["x-agent-token"]
|
||||
if (typeof headerToken === "string" && headerToken.length > 0) return headerToken
|
||||
|
||||
const authHeader = req.headers.authorization
|
||||
if (authHeader?.startsWith("Bearer ")) return authHeader.slice(7)
|
||||
|
||||
return null
|
||||
}
|
||||
|
||||
const pickFileTargets = (target: unknown) => {
|
||||
if (!target || typeof target !== "object" || Array.isArray(target)) return {}
|
||||
|
||||
const allowedFields = [
|
||||
"project",
|
||||
"customer",
|
||||
"contract",
|
||||
"vendor",
|
||||
"incominginvoice",
|
||||
"plant",
|
||||
"createddocument",
|
||||
"vehicle",
|
||||
"product",
|
||||
"check",
|
||||
"inventoryitem",
|
||||
"space",
|
||||
"documentbox",
|
||||
"authProfile",
|
||||
]
|
||||
|
||||
return Object.fromEntries(
|
||||
Object.entries(target as Record<string, any>)
|
||||
.filter(([key, value]) => allowedFields.includes(key) && value !== undefined && value !== null)
|
||||
)
|
||||
}
|
||||
|
||||
export default async function instanceAgentGatewayRoutes(server: FastifyInstance) {
|
||||
await server.register(multipart, {
|
||||
limits: { fileSize: 100 * 1024 * 1024 },
|
||||
})
|
||||
|
||||
const authenticateAgent = async (req: FastifyRequest, reply: any) => {
|
||||
const token = readAgentToken(req)
|
||||
if (!token) {
|
||||
reply.code(401).send({ error: "Agent token required" })
|
||||
return null
|
||||
}
|
||||
|
||||
const [agent] = await server.db
|
||||
.select()
|
||||
.from(instanceAgents)
|
||||
.where(and(
|
||||
eq(instanceAgents.tokenHash, hashToken(token)),
|
||||
eq(instanceAgents.active, true)
|
||||
))
|
||||
.limit(1)
|
||||
|
||||
if (!agent) {
|
||||
reply.code(401).send({ error: "Invalid agent token" })
|
||||
return null
|
||||
}
|
||||
|
||||
return agent
|
||||
}
|
||||
|
||||
server.post("/heartbeat", async (req, reply) => {
|
||||
const agent = await authenticateAgent(req, reply)
|
||||
if (!agent) return
|
||||
|
||||
const body = (req.body || {}) as {
|
||||
capabilities?: Record<string, any>
|
||||
scannerNames?: string[]
|
||||
printerNames?: string[]
|
||||
debugInfo?: Record<string, any>
|
||||
}
|
||||
|
||||
await server.db
|
||||
.update(instanceAgents)
|
||||
.set({
|
||||
capabilities: body.capabilities || agent.capabilities,
|
||||
scannerNames: body.scannerNames || agent.scannerNames,
|
||||
printerNames: body.printerNames || agent.printerNames,
|
||||
lastDebugInfo: body.debugInfo || null,
|
||||
lastSeenAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(instanceAgents.id, agent.id))
|
||||
|
||||
const [pending] = await server.db
|
||||
.select({ count: sql<number>`count(*)::int` })
|
||||
.from(instanceAgentScanJobs)
|
||||
.where(and(
|
||||
eq(instanceAgentScanJobs.agentId, agent.id),
|
||||
eq(instanceAgentScanJobs.status, "pending")
|
||||
))
|
||||
|
||||
return {
|
||||
status: "ok",
|
||||
pendingScanJobs: pending?.count || 0,
|
||||
}
|
||||
})
|
||||
|
||||
server.get("/scan-jobs/next", async (req, reply) => {
|
||||
const agent = await authenticateAgent(req, reply)
|
||||
if (!agent) return
|
||||
|
||||
const [pendingJob] = await server.db
|
||||
.select()
|
||||
.from(instanceAgentScanJobs)
|
||||
.where(and(
|
||||
eq(instanceAgentScanJobs.agentId, agent.id),
|
||||
eq(instanceAgentScanJobs.status, "pending")
|
||||
))
|
||||
.orderBy(asc(instanceAgentScanJobs.createdAt))
|
||||
.limit(1)
|
||||
|
||||
if (!pendingJob) return { job: null }
|
||||
|
||||
const [claimedJob] = await server.db
|
||||
.update(instanceAgentScanJobs)
|
||||
.set({
|
||||
status: "running",
|
||||
claimedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
attempts: pendingJob.attempts + 1,
|
||||
})
|
||||
.where(and(
|
||||
eq(instanceAgentScanJobs.id, pendingJob.id),
|
||||
eq(instanceAgentScanJobs.status, "pending")
|
||||
))
|
||||
.returning()
|
||||
|
||||
return { job: claimedJob || null }
|
||||
})
|
||||
|
||||
server.post<{ Params: { id: string } }>("/scan-jobs/:id/status", async (req, reply) => {
|
||||
const agent = await authenticateAgent(req, reply)
|
||||
if (!agent) return
|
||||
|
||||
const body = (req.body || {}) as { status?: string; message?: string }
|
||||
const allowedStatuses = ["running", "failed", "canceled"]
|
||||
|
||||
if (!body.status || !allowedStatuses.includes(body.status)) {
|
||||
return reply.code(400).send({ error: "Invalid status" })
|
||||
}
|
||||
|
||||
const [job] = await server.db
|
||||
.update(instanceAgentScanJobs)
|
||||
.set({
|
||||
status: body.status,
|
||||
agentMessage: body.message,
|
||||
finishedAt: ["failed", "canceled"].includes(body.status) ? new Date() : undefined,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(and(
|
||||
eq(instanceAgentScanJobs.id, req.params.id),
|
||||
eq(instanceAgentScanJobs.agentId, agent.id)
|
||||
))
|
||||
.returning()
|
||||
|
||||
if (!job) return reply.code(404).send({ error: "Scan job not found" })
|
||||
|
||||
return { job }
|
||||
})
|
||||
|
||||
server.post<{ Params: { id: string } }>("/scan-jobs/:id/upload", async (req, reply) => {
|
||||
const agent = await authenticateAgent(req, reply)
|
||||
if (!agent) return
|
||||
|
||||
const [job] = await server.db
|
||||
.select()
|
||||
.from(instanceAgentScanJobs)
|
||||
.where(and(
|
||||
eq(instanceAgentScanJobs.id, req.params.id),
|
||||
eq(instanceAgentScanJobs.agentId, agent.id)
|
||||
))
|
||||
.limit(1)
|
||||
|
||||
if (!job) return reply.code(404).send({ error: "Scan job not found" })
|
||||
if (!["running", "pending"].includes(job.status)) {
|
||||
return reply.code(409).send({ error: "Scan job is not uploadable" })
|
||||
}
|
||||
|
||||
const data: any = await req.file()
|
||||
if (!data?.file) return reply.code(400).send({ error: "No file uploaded" })
|
||||
|
||||
const fileBuffer = await data.toBuffer()
|
||||
const filename = job.requestedFilename || data.filename || `${job.id}.pdf`
|
||||
|
||||
const createdFile = await saveFile(
|
||||
server,
|
||||
job.tenantId,
|
||||
null,
|
||||
{
|
||||
filename,
|
||||
content: fileBuffer,
|
||||
contentType: data.mimetype || "application/pdf",
|
||||
},
|
||||
null,
|
||||
null,
|
||||
{
|
||||
...pickFileTargets(job.target),
|
||||
createdBy: job.requestedBy,
|
||||
}
|
||||
)
|
||||
|
||||
if (!createdFile) return reply.code(500).send({ error: "Could not save scan file" })
|
||||
|
||||
const [updatedJob] = await server.db
|
||||
.update(instanceAgentScanJobs)
|
||||
.set({
|
||||
status: "completed",
|
||||
fileId: createdFile.id,
|
||||
finishedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(instanceAgentScanJobs.id, job.id))
|
||||
.returning()
|
||||
|
||||
return {
|
||||
job: updatedJob,
|
||||
file: createdFile,
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user