SnapOtter/tests/integration/process-contract.test.ts

/**
 * Integration tests for the ref-based v2 process contract.
 *
 * Registers a native-v2 tool that receives TWO input refs and writes
 * a scratch-dir output (concatenation of both inputs). Verifies the
 * worker loads all refs, invokes processV2 (not the legacy path), and
 * resolves scratchPath outputs correctly.
 */
import { randomUUID } from "node:crypto";
import { writeFile } from "node:fs/promises";
import { join } from "node:path";
import { eq } from "drizzle-orm";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { db, schema } from "../../apps/api/src/db/index.js";
import { enqueueToolJob, waitForJob } from "../../apps/api/src/jobs/enqueue.js";
import type { ToolJobData } from "../../apps/api/src/jobs/types.js";
import { getObjectBuffer, putObject } from "../../apps/api/src/lib/object-storage.js";
import { registerToolProcessFn } from "../../apps/api/src/routes/tool-factory.js";
import { buildTestApp, type TestApp } from "./test-server.js";

// Register a native-v2 test tool that concatenates two inputs via scratchPath
registerToolProcessFn({
  toolId: "contract-v2",
  settingsSchema: { parse: (v: unknown) => v } as never,
  process: async () => {
    throw new Error("legacy path must not run");
  },
  processV2: async (ctx) => {
    if (ctx.inputs.length !== 2) throw new Error(`expected 2 inputs, got ${ctx.inputs.length}`);
    const outPath = join(ctx.scratchDir, "combined.txt");
    await writeFile(outPath, Buffer.concat(ctx.inputs.map((i) => i.buffer)));
    return { scratchPath: outPath, filename: "combined.txt", contentType: "text/plain" };
  },
});

// Register a legacy-only test tool (no processV2) to verify the adapter
registerToolProcessFn({
  toolId: "contract-legacy-echo",
  settingsSchema: { parse: (v: unknown) => v } as never,
  process: async (inputBuffer: Buffer, _settings: unknown, filename: string) => {
    return { buffer: inputBuffer, filename, contentType: "application/octet-stream" };
  },
});

let testApp: TestApp;

beforeAll(async () => {
  testApp = await buildTestApp();
}, 30_000);

afterAll(async () => {
  await testApp.cleanup();
}, 10_000);

describe("V2 process contract", () => {
  it("loads all inputRefs and invokes processV2 with scratchPath output", async () => {
    const jobId = randomUUID();
    const bufA = Buffer.from("AAAA");
    const bufB = Buffer.from("BBBB");

    // Store two inputs in object storage
    const refA = `uploads/${jobId}/input-a.txt`;
    const refB = `uploads/${jobId}/input-b.txt`;
    await putObject(refA, bufA);
    await putObject(refB, bufB);

    const data: ToolJobData = {
      jobId,
      toolId: "contract-v2",
      userId: null,
      pool: "image",
      inputRefs: [refA, refB],
      filename: "input-a.txt",
      settings: {},
      kind: "tool",
    };

    await enqueueToolJob(data);

    const result = await waitForJob("image", jobId, 10_000);
    expect(result).not.toBeNull();
    expect(result!.filename).toBe("combined.txt");
    expect(result!.outputRefs.length).toBeGreaterThan(0);

    // Verify the output is the concatenation of both inputs
    const outputBuffer = await getObjectBuffer(result!.outputRefs[0]);
    expect(outputBuffer.toString()).toBe("AAAABBBB");

    // Verify sizes: original is the primary input, processed is the concatenated output
    expect(result!.originalSize).toBe(4); // bufA.length
    expect(result!.processedSize).toBe(8); // bufA.length + bufB.length

    // Verify the durable DB row
    const [row] = await db.select().from(schema.jobs).where(eq(schema.jobs.id, jobId));
    expect(row).toBeDefined();
    expect(row!.status).toBe("completed");
    expect(row!.bytesIn).toBe(4);
    expect(row!.bytesOut).toBe(8);
    expect(row!.outputRefs).toBeDefined();
    expect((row!.outputRefs as string[]).length).toBeGreaterThan(0);
  });

  it("passes input refs and filenames correctly to processV2", async () => {
    const jobId = randomUUID();
    const bufA = Buffer.from("first");
    const bufB = Buffer.from("second");

    const refA = `uploads/${jobId}/primary.bin`;
    const refB = `uploads/${jobId}/secondary.bin`;
    await putObject(refA, bufA);
    await putObject(refB, bufB);

    // Register a tool that returns metadata about what it received
    registerToolProcessFn({
      toolId: "contract-v2-meta",
      settingsSchema: { parse: (v: unknown) => v } as never,
      process: async () => {
        throw new Error("legacy path must not run");
      },
      processV2: async (ctx) => {
        const meta = {
          count: ctx.inputs.length,
          refs: ctx.inputs.map((i) => i.ref),
          filenames: ctx.inputs.map((i) => i.filename),
          sizes: ctx.inputs.map((i) => i.buffer.length),
        };
        return {
          buffer: Buffer.from(JSON.stringify(meta)),
          filename: "meta.json",
          contentType: "application/json",
          resultPayload: meta,
        };
      },
    });

    const data: ToolJobData = {
      jobId,
      toolId: "contract-v2-meta",
      userId: null,
      pool: "image",
      inputRefs: [refA, refB],
      filename: "my-file.bin",
      settings: {},
      kind: "tool",
    };

    await enqueueToolJob(data);

    const result = await waitForJob("image", jobId, 10_000);
    expect(result).not.toBeNull();

    // Primary filename is data.filename, secondary derives from the ref basename
    expect(result!.resultPayload).toBeDefined();
    const meta = result!.resultPayload as {
      count: number;
      refs: string[];
      filenames: string[];
      sizes: number[];
    };
    expect(meta.count).toBe(2);
    expect(meta.refs).toEqual([refA, refB]);
    expect(meta.filenames[0]).toBe("my-file.bin"); // primary keeps client name
    expect(meta.filenames[1]).toBe("secondary.bin"); // derived from ref basename
    expect(meta.sizes).toEqual([5, 6]);
  });

  it("legacy adapter routes single-input tools through processV2", async () => {
    const jobId = randomUUID();
    const buf = Buffer.from("echo-me");

    const ref = `uploads/${jobId}/test.bin`;
    await putObject(ref, buf);

    // contract-legacy-echo was registered with only a legacy process
    // function. registerToolProcessFn should have wrapped it via
    // adaptLegacyProcess so the worker calls processV2 internally.
    const data: ToolJobData = {
      jobId,
      toolId: "contract-legacy-echo",
      userId: null,
      pool: "image",
      inputRefs: [ref],
      filename: "test.bin",
      settings: {},
      kind: "tool",
    };

    await enqueueToolJob(data);

    const result = await waitForJob("image", jobId, 10_000);
    expect(result).not.toBeNull();
    const outputBuffer = await getObjectBuffer(result!.outputRefs[0]);
    expect(outputBuffer.toString()).toBe("echo-me");
  });
});