diff --git a/.gitignore b/.gitignore index a98ff79..32f3742 100644 --- a/.gitignore +++ b/.gitignore @@ -20,7 +20,6 @@ build/ .vercel *.tsbuildinfo next-env.d.ts - -# Storage directories -storage/screenshots/ +data/ storage/processed/ +storage/screenshots/ diff --git a/src/app/api/screenshots/[jobId]/route.ts b/src/app/api/screenshots/[jobId]/route.ts index 53dfa64..1472b21 100644 --- a/src/app/api/screenshots/[jobId]/route.ts +++ b/src/app/api/screenshots/[jobId]/route.ts @@ -1,22 +1,19 @@ import { NextRequest, NextResponse } from "next/server"; import { getServerRuntime } from "@/server/runtime"; +import { toJobView } from "@/server/types/queue"; export async function GET( - _req: NextRequest, - { params }: { params: Promise<{ jobId: string }>} + _req: NextRequest, + { params }: { params: Promise<{ jobId: string }> }, ) { - const { jobId } = await params; - const { queue } = await getServerRuntime(); + const { jobId } = await params; + const { queue } = await getServerRuntime(); - const job = queue.getById(jobId); + const job = queue.getById(jobId); - if(!job) { - return NextResponse.json({ error: "Job not found"}, { status: 404 }); - } + if (!job) { + return NextResponse.json({ error: "Job not found" }, { status: 404 }); + } - return NextResponse.json({ - jobId: job.id, - status: "queued", - createdAt: job.createdAt, - }); + return NextResponse.json(toJobView(job)); } \ No newline at end of file diff --git a/src/app/api/screenshots/route.ts b/src/app/api/screenshots/route.ts index 930f0c4..169b118 100644 --- a/src/app/api/screenshots/route.ts +++ b/src/app/api/screenshots/route.ts @@ -1,76 +1,52 @@ import { NextRequest, NextResponse } from "next/server"; import { getServerRuntime } from "@/server/runtime"; +import type { JobStatus } from "@/server/types/queue"; import { randomUUID } from "crypto"; import { writeFile, mkdir } from "fs/promises"; import path from "path"; import { env } from "@/server/config/env"; import type { ScreenshotInput } from "@/server/types/screenshot"; -const ACCEPTED_MIME_TYPES = [ - "image/png", - "image/jpeg", - "image/webp", - "image/heic", - "image/gif", - "image/bmp", -]; +function parseStatusFilter(url: string): JobStatus[] | undefined { + const { searchParams } = new URL(url, "http://localhost"); + const raw = searchParams.get("status"); + if (!raw) return undefined; -const ACCEPTED_EXTENSIONS = /\.(png|jpe?g|webp|heic|gif|bmp)$/i; -const MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024; + const valid: JobStatus[] = ["queued", "processing", "processed", "failed"]; + const parsed = raw + .split(",") + .map((s) => s.trim()) + .filter((s): s is JobStatus => valid.includes(s as JobStatus)); -export async function GET() { - const { repository } = await getServerRuntime(); - const records = await repository.findAll(); - return NextResponse.json(records); + return parsed.length > 0 ? parsed : undefined; +} + +export async function GET(req: NextRequest) { + const { queue } = await getServerRuntime(); + const filter = parseStatusFilter(req.url); + return NextResponse.json(queue.listViews(filter)); } export async function POST(req: NextRequest) { - try{ + try { const formData = await req.formData(); - const file = formData.get("file") as File | null; - const sourceType = formData.get("sourceType") as string | null; - const sourceRef = formData.get("sourceRef") as string | null; - const description = formData.get("description") as string | null; + const file = formData.get("file") as File | null; + const sourceType = formData.get("sourceType") as string | null; + const sourceRef = formData.get("sourceRef") as string | null; + const description = formData.get("description") as string | null; const originalFileName = formData.get("originalFileName") as string | null; - if(!file || !sourceType ) { + if (!file || !sourceType) { return NextResponse.json( - { error: "file and sourceType are required"}, - { status: 400 } + { error: "file and sourceType are required" }, + { status: 400 }, ); } - if (!["telegram", "local", "cloud"].includes(sourceType)) { - return NextResponse.json( - { error: "sourceType must be telegram, local, or cloud"}, - { status: 400 } - ); - } - - if( - !ACCEPTED_MIME_TYPES.includes(file.type) || - !ACCEPTED_EXTENSIONS.test(file.name) - ) { - return NextResponse.json( - { error: - "Only image files are accepted (png, jpg, webp, heic, gif, bmp)", - }, - { status: 400 } - ); - } - - if (file.size > MAX_FILE_SIZE_BYTES) { - return NextResponse.json( - { error: "File size must not exceed 10MB" }, - { status: 400 } - ); - } - - const id = randomUUID(); - const originalName = originalFileName ?? file.name; - const ext = path.extname(originalName) || ".png"; - const fileName = `${id}_${path.basename(originalName, ext)}${ext}`; + const id = randomUUID(); + const ext = path.extname(file.name) || ".png"; + const fileName = `${id}${ext}`; const storagePath = path.join(env.screenshotStorageDir, fileName); await mkdir(env.screenshotStorageDir, { recursive: true }); @@ -87,28 +63,28 @@ export async function POST(req: NextRequest) { const input: ScreenshotInput = { id, - sourceType: sourceType as ScreenshotInput["sourceType"], - sourceRef: sourceRef ?? originalName, - storagePath, - createdAt: new Date().toISOString(), + sourceType: sourceType as ScreenshotInput["sourceType"], + sourceRef: sourceRef ?? fileName, + filePath: storagePath, + createdAt: new Date().toISOString(), metadata: { - originalFileName: originalName, - description: description ?? undefined, + originalFileName: originalFileName ?? file.name, + description: description ?? undefined, }, }; - const { queue, workerTrigger } = await getServerRuntime(); - await queue.enqueue("process-screenshot", input); - - // Trigger the worker without awaiting it - void workerTrigger(); + const { queue } = await getServerRuntime(); + const job = await queue.enqueue("process-screenshot", input); - return NextResponse.json({ jobId: id, status: "queued" }, { status: 202 }); + return NextResponse.json( + { jobId: job.id, status: job.status }, + { status: 202 }, + ); } catch (err) { console.error("[POST /api/screenshots]", err); return NextResponse.json( { error: "Internal server error" }, - { status: 500 } + { status: 500 }, ); } } \ No newline at end of file diff --git a/src/server/runtime.ts b/src/server/runtime.ts index d7de10a..a9377a5 100644 --- a/src/server/runtime.ts +++ b/src/server/runtime.ts @@ -37,11 +37,10 @@ const storageReady = processedStorage.ensure(); export async function getServerRuntime() { await storageReady; - return { pipeline, queue, repository, workerTrigger: worker.trigger, }; -} +} \ No newline at end of file diff --git a/src/server/services/queue/queue.ts b/src/server/services/queue/queue.ts index 94dc7af..7aade10 100644 --- a/src/server/services/queue/queue.ts +++ b/src/server/services/queue/queue.ts @@ -1,51 +1,120 @@ import { createId } from "../../utils/id"; -import type { QueueJob } from "../../types/queue"; +import { + isValidTransition, + toJobView, + type Job, + type JobStatus, + type JobView, + type PipelineStage, +} from "../../types/queue"; +import type { ScreenshotInput } from "../../types/screenshot"; -type JobHandler = (job: QueueJob) => Promise; +type JobHandler = (job: Job) => Promise; -export class InMemoryQueue { - private readonly jobs: Map> = new Map(); - private readonly queue: string[] = []; +export class InMemoryQueue { + private readonly store = new Map(); + private readonly order: string[] = []; - async enqueue(name: string, payload: TPayload): Promise> { - const job: QueueJob = { - id: createId("job"), + async enqueue(name: string, payload: TPayload): Promise { + const job: Job = { + id: createId("job"), name, payload, - createdAt: new Date().toISOString(), - status: "queued", + status: "queued", + stage: null, + createdAt: new Date(), + stageErrors: {}, }; - this.jobs.set(job.id, job); - this.queue.push(job.id); + this.store.set(job.id, job); + this.order.push(job.id); return job; } - async process(handler: JobHandler): Promise { - while (this.queue.length > 0) { - const id = this.queue.shift(); - if (!id) { - return; - } + updateStatus( + id: string, + to: JobStatus, + stage: PipelineStage = null, + error?: string, + ): void { + const job = this.store.get(id); + if (!job) return; + + if (!isValidTransition(job.status, to)) { + console.warn( + `[Queue] Blocked invalid transition ${job.status} → ${to} for job ${id}`, + ); + return; + } + + job.status = to; + job.stage = stage; + + if (to === "processing" && !job.startedAt) { + job.startedAt = new Date(); + } - const job = this.jobs.get(id); - if (!job) { - continue; + if (to === "processed" || to === "failed") { + job.completedAt = new Date(); + } + + if (error) { + if (stage) { + job.stageErrors[stage] = error; } + job.error = error; + } + } + + retry(id: string): boolean { + const job = this.store.get(id); + if (!job) return false; + if (!isValidTransition(job.status, "queued")) { + return false; + } + + job.status = "queued"; + job.stage = null; + job.error = undefined; + job.startedAt = undefined; + job.completedAt = undefined; + job.stageErrors = {}; + this.order.push(id); + return true; + } + + async process(handler: JobHandler): Promise { + while (this.order.length > 0) { + const id = this.order.shift(); + if (!id) continue; + const job = this.store.get(id); + if (!job || job.status === "processed") continue; await handler(job); } } - size(): number { - return this.queue.length; + getById(id: string): Job | undefined { + return this.store.get(id); } - getById(id: string): QueueJob | undefined { - return this.jobs.get(id); + list(statusFilter?: JobStatus[]): Job[] { + const all = Array.from(this.store.values()).sort( + (a, b) => b.createdAt.getTime() - a.createdAt.getTime(), + ); + + if (statusFilter && statusFilter.length > 0) { + return all.filter((j) => statusFilter.includes(j.status)); + } + + return all; } - list(): QueueJob[] { - return Array.from(this.jobs.values()); + listViews(statusFilter?: JobStatus[]): JobView[] { + return this.list(statusFilter).map(toJobView); + } + + size(): number { + return this.order.length; } } diff --git a/src/server/types/queue.ts b/src/server/types/queue.ts index 48f13da..56c0dfd 100644 --- a/src/server/types/queue.ts +++ b/src/server/types/queue.ts @@ -2,6 +2,64 @@ import type { ScreenshotInput } from "./screenshot"; export type JobStatus = "queued" | "processing" | "processed" | "failed"; +export type PipelineStage = + | "ocr" + | "vision" + | "source" + | "tagging" + | "storing" + | null; + +const VALID_TRANSITIONS: Record = { + queued: ["processing"], + processing: ["processed", "failed"], + processed: [], + failed: ["queued"], +}; + +export function isValidTransition(from: JobStatus, to: JobStatus): boolean { + return VALID_TRANSITIONS[from].includes(to); +} + +export interface Job { + id: string; + name: string; + payload: ScreenshotInput; + status: JobStatus; + stage: PipelineStage; + createdAt: Date; + startedAt?: Date; + completedAt?: Date; + error?: string; + stageErrors: Record; +} + +export interface JobView { + id: string; + name: string; + status: JobStatus; + stage: PipelineStage; + error?: string; + stageErrors: Record; + createdAt: string; + startedAt?: string; + completedAt?: string; +} + +export function toJobView(job: Job): JobView { + return { + id: job.id, + name: job.name, + status: job.status, + stage: job.stage, + error: job.error, + stageErrors: job.stageErrors, + createdAt: job.createdAt.toISOString(), + startedAt: job.startedAt?.toISOString(), + completedAt: job.completedAt?.toISOString(), + }; +} + export interface QueueJob { id: string; name: string; @@ -12,4 +70,4 @@ export interface QueueJob { error?: string; } -export type ScreenshotJob = QueueJob; +export type ScreenshotJob = QueueJob; \ No newline at end of file