Feature/job state model

.gitignore

@@ -20,7 +20,6 @@ build/
 .vercel
 *.tsbuildinfo
 next-env.d.ts
-
-# Storage directories
-storage/screenshots/
+data/
 storage/processed/
+storage/screenshots/

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));
 }

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 },
     );
   }
 }

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,
   };
-}
+}

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<TPayload> = (job: QueueJob<TPayload>) => Promise<void>;
+type JobHandler = (job: Job) => Promise<void>;
 
-export class InMemoryQueue<TPayload> {
-  private readonly jobs: Map<string, QueueJob<TPayload>> = new Map();
-  private readonly queue: string[] = [];
+export class InMemoryQueue<TPayload extends ScreenshotInput = ScreenshotInput> {
+  private readonly store = new Map<string, Job>();
+  private readonly order: string[] = [];
 
-  async enqueue(name: string, payload: TPayload): Promise<QueueJob<TPayload>> {
-    const job: QueueJob<TPayload> = {
-      id: createId("job"),
+  async enqueue(name: string, payload: TPayload): Promise<Job> {
+    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<TPayload>): Promise<void> {
-    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<void> {
+    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<TPayload> | 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<TPayload>[] {
-    return Array.from(this.jobs.values());
+  listViews(statusFilter?: JobStatus[]): JobView[] {
+    return this.list(statusFilter).map(toJobView);
+  }
+
+  size(): number {
+    return this.order.length;
   }
 }