import {

  computeAggregates,

  computeComparisons,

  effectiveSubject,

  effectiveSubjectKind,

  rowsToMeasurements,

  scorecardFromCells,

  sha256Hex,

  stableId,

  trendSeries,

  upsertSubjects,

  verifySignature,

} from "./reporting.mjs";

// Re-export the pure ingest/reporting functions so tests (node --test) and

// any future consumers can import them straight from the worker module.

export {

  computeAggregates,

  computeComparisons,

  effectiveSubject,

  effectiveSubjectKind,

  rowsToMeasurements,

  scorecardFromCells,

  trendSeries,

  upsertSubjects,

  verifySignature,

};

export const MAX_UPLOAD_BYTES = 25 * 1024 * 1024;

export default {

  async fetch(request, env) {

    const url = new URL(request.url);

    const { pathname } = url;

    if (request.method === "GET") {

      if (pathname === "/health") {

        return json({ ok: true, service: "oak-benchmark-upload" });

      }

      if (pathname === "/api/scorecard") {

        return getScorecard(env);

      }

      if (pathname === "/api/trends") {

        return getTrends(env, url);

      }

      if (pathname.startsWith("/api/baseline-book/")) {

        return getBaselineBook(env, decodeURIComponent(pathname.slice("/api/baseline-book/".length)));

      }

      if (pathname === "/api/verifications") {

        return getVerifications(env);

      }

      return json({ ok: false, error: "not_found" }, 404);

    }

    if (request.method !== "POST") {

      return json({ ok: false, error: "not_found" }, 404);

    }

    if (pathname === "/webhook/github") {

      return handleGithubWebhook(request, env);

    }

    const authError = authorize(request, env);

    if (authError) {

      return authError;

    }

    if (pathname === "/api/scorecard") {

      return postScorecard(request, env);

    }

    if (pathname === "/api/baseline-book") {

      return postBaselineBook(request, env);

    }

    if (pathname === "/api/verifications") {

      return postVerification(request, env);

    }

    if (["/", "/upload", "/api/upload"].includes(pathname)) {

      return handleUpload(request, env);

    }

    return json({ ok: false, error: "not_found" }, 404);

  },

};

// --- Upload + ingest --------------------------------------------------------

async function handleUpload(request, env) {

  const contentLength = Number(request.headers.get("content-length") || "0");

  if (contentLength > MAX_UPLOAD_BYTES) {

    return json({ ok: false, error: "upload_too_large", max_bytes: MAX_UPLOAD_BYTES }, 413);

  }

  const body = await request.text();

  const bodyBytes = new TextEncoder().encode(body);

  if (bodyBytes.byteLength > MAX_UPLOAD_BYTES) {

    return json({ ok: false, error: "upload_too_large", max_bytes: MAX_UPLOAD_BYTES }, 413);

  }

  let parsed;

  try {

    parsed = parseJsonl(body);

  } catch (error) {

    return json({ ok: false, error: "invalid_jsonl", detail: String(error.message || error) }, 400);

  }

  const firstRow = parsed.firstRow;

  const benchId = stringValue(firstRow.bench_id);

  const profile = stringValue(firstRow.profile);

  const timestampUtc = stringValue(firstRow.timestamp_utc || firstRow.bench_id);

  if (!benchId || !profile) {

    return json({ ok: false, error: "missing_required_metadata", required: ["bench_id", "profile"] }, 400);

  }

  const sha256 = await sha256Hex(body);

  const r2Key = objectKey(profile, timestampUtc, benchId, sha256);

  const now = new Date().toISOString();

  const persistence = {

    r2: "not_configured",

    d1: "not_configured",

  };

  if (env.OAK_BENCH_RESULTS && typeof env.OAK_BENCH_RESULTS.put === "function") {

    await env.OAK_BENCH_RESULTS.put(r2Key, body, {

      httpMetadata: { contentType: "application/x-ndjson; charset=utf-8" },

      customMetadata: {

        bench_id: benchId,

        profile,

        timestamp_utc: timestampUtc,

        sha256,

        row_count: String(parsed.rowCount),

        source_oak_hash: stringValue(firstRow.source && firstRow.source.oak_hash) || "",

        host: stringValue(firstRow.host) || "",

        machine: stringValue(firstRow.machine) || "",

      },

    });

    persistence.r2 = "stored";

  }

  if (env.OAK_BENCH_DB && typeof env.OAK_BENCH_DB.prepare === "function") {

    await persistUploadMetadata(env.OAK_BENCH_DB, {

      firstRow,

      r2Key,

      sha256,

      rowCount: parsed.rowCount,

      byteCount: bodyBytes.byteLength,

      contentType: request.headers.get("content-type") || "application/x-ndjson",

      now,

    });

    persistence.d1 = "upload_metadata_recorded";

    const runKey = `${benchId}:${profile}`;

    const ingested = await ingestRows(env.OAK_BENCH_DB, parsed.rows, runKey, profile, now);

    persistence.d1 = "ingested";

    persistence.measurements = ingested.measurements;

    persistence.subjects = ingested.subjects;

    persistence.aggregates = ingested.aggregates;

    persistence.comparisons = ingested.comparisons;

    persistence.skipped_no_subject = ingested.skipped_no_subject;

  }

  return json({

    ok: true,

    bench_id: benchId,

    profile,

    row_count: parsed.rowCount,

    sha256,

    r2_key: r2Key,

    persistence,

  });

}

/**

 * The four ingest steps, all computed by pure functions in reporting.mjs:

 * 1. Upsert run-scoped subjects keyed (subject, subject_kind, binary_sha256).

 * 2. Insert each JSONL row into measurements (null-preserving projection).

 * 3. Per-group p50/p90/p95/p99 nearest-rank latency aggregates + averages.

 * 4. Target-vs-git comparison rows (delta/slower_pct/ratio per reporting.py).

 *

 * Exported so tests can drive the full ingest flow against a stub D1

 * binding (see reporting_parity.test.mjs re-upload test).

 */

export async function ingestRows(db, rows, runKey, profile, now) {

  const firstRow = rows[0] || {};

  const subjectDetails = firstRow.subject_details && typeof firstRow.subject_details === "object"

    ? firstRow.subject_details

    : {};

  const subjectVersions = firstRow.subject_versions && typeof firstRow.subject_versions === "object"

    ? firstRow.subject_versions

    : {};

  // Step 1: subjects. Identity is the deterministic subject_id, which

  // stableId-encodes (run_key, effective subject, kind, binary_sha256): two

  // binary shas for one subject name in a run stay two distinct subject

  // records (schema_v2.sql documents that schema.sql's inline

  // UNIQUE(run_key, name) is superseded by subject_id identity). Re-uploads

  // hash to the same id and take the conflict path, so no read-back remap is

  // needed. Rows with no subject identity (no subject AND no platform) are

  // counted in skipped_no_subject, never silently dropped.

  const { records: subjects, idsByKey, skipped } = await upsertSubjects(rows, runKey, subjectDetails, subjectVersions);

  await batched(db, subjects.map((subject) => db.prepare(`

    INSERT INTO subjects (subject_id, run_key, name, kind, label, version, binary_path, metadata_json)

    VALUES (?, ?, ?, ?, ?, ?, ?, ?)

    ON CONFLICT(subject_id) DO UPDATE SET

      name = excluded.name,

      kind = excluded.kind,

      label = excluded.label,

      version = excluded.version,

      binary_path = excluded.binary_path,

      metadata_json = excluded.metadata_json

  `).bind(

    subject.subject_id,

    subject.run_key,

    subject.name,

    subject.kind,

    subject.label,

    subject.version,

    subject.binary_path,

    subject.metadata_json,

  )));

  // Step 2: measurements (idempotent re-upload: replace this run's rows).

  // Derived tables (aggregates, comparisons) are deleted alongside the

  // measurements: they are recomputed from this upload's measurements, so a

  // re-upload with FEWER groups must not leave stale rows from a previous

  // upload behind (upserts alone only overwrite groups that still exist).

  const ingestReport = { skipped_no_subject: skipped };

  const measurements = rowsToMeasurements(rows, runKey, now, idsByKey, ingestReport);

  const aggregates = computeAggregates(measurements, profile, now);

  const comparisons = computeComparisons(aggregates);

  // Keep the destructive run rewrite in one D1 batch. D1 batches are

  // transactional, so a mid-ingest insert failure must not commit only the

  // deletes and leave the run empty or partially rebuilt. Subject upserts

  // above are non-destructive and can safely remain outside this rewrite.

  const runRewriteStatements = [

    db.prepare("DELETE FROM measurements WHERE run_key = ?").bind(runKey),

    db.prepare("DELETE FROM aggregates WHERE run_key = ?").bind(runKey),

    db.prepare("DELETE FROM comparisons WHERE run_key = ?").bind(runKey),

    ...measurements.map((m) => db.prepare(`

    INSERT INTO measurements (

      run_key, subject_id, scenario, operation, run_index, is_canonical,

      elapsed_ms, estimated_tokens_total, estimated_tokens_input, estimated_tokens_output,

      token_source, raw_output_bytes, stdout_bytes, stderr_bytes, output_truncated,

      tool_call_count, vcs_tool_call_count, terminal_tool_call_count, returncode,

      command_json, stderr, row_json, created_at,

      runner_id, runner_class, load_tier, netshape_profile

    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

  `).bind(

    m.run_key, m.subject_id, m.scenario, m.operation, m.run_index, m.is_canonical,

    m.elapsed_ms, m.estimated_tokens_total, m.estimated_tokens_input, m.estimated_tokens_output,

    m.token_source, m.raw_output_bytes, m.stdout_bytes, m.stderr_bytes, m.output_truncated,

    m.tool_call_count, m.vcs_tool_call_count, m.terminal_tool_call_count, m.returncode,

    m.command_json, m.stderr, m.row_json, m.created_at,

    m.runner_id, m.runner_class, m.load_tier, m.netshape_profile,

  )),

    // Step 3: aggregates (plain INSERTs: the run's rows are deleted in the

    // same batch, so there is nothing to conflict with after commit).

    ...aggregates.map((a) => db.prepare(`

    INSERT INTO aggregates (

      run_key, subject_id, profile, scenario, operation, is_canonical,

      sample_count, failure_count, failure_rate,

      p50_ms, p90_ms, p95_ms, p99_ms, mean_ms, min_ms, max_ms,

      avg_tokens, avg_output_bytes, truncated_count, avg_tool_calls, updated_at

    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

  `).bind(

    a.run_key, a.subject_id, a.profile, a.scenario, a.operation, a.is_canonical,

    a.sample_count, a.failure_count, a.failure_rate,

    a.p50_ms, a.p90_ms, a.p95_ms, a.p99_ms, a.mean_ms, a.min_ms, a.max_ms,

    a.avg_tokens, a.avg_output_bytes, a.truncated_count, a.avg_tool_calls, a.updated_at,

  )),

    // Step 4: comparisons (plain INSERTs, same reasoning as step 3).

    ...comparisons.map((c) => db.prepare(`

    INSERT INTO comparisons (

      run_key, target_subject_id, baseline_subject_id, baseline_kind, profile,

      scenario, operation,

      p50_delta_ms, p50_slower_pct, p90_delta_ms, p90_slower_pct,

      p99_delta_ms, p99_slower_pct,

      avg_token_delta, token_higher_pct, avg_output_byte_delta, output_byte_higher_pct,

      avg_tool_call_delta, tool_call_higher_pct, failure_rate_delta_pp,

      threshold_pct, min_delta_ms, is_regression, regression_dimensions, state, severity_score, updated_at

    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

  `).bind(

    c.run_key, c.target_subject_id, c.baseline_subject_id, c.baseline_kind, c.profile,

    c.scenario, c.operation,

    c.p50_delta_ms, c.p50_slower_pct, c.p90_delta_ms, c.p90_slower_pct,

    c.p99_delta_ms, c.p99_slower_pct,

    c.avg_token_delta, c.token_higher_pct, c.avg_output_byte_delta, c.output_byte_higher_pct,

    c.avg_tool_call_delta, c.tool_call_higher_pct, c.failure_rate_delta_pp,

    c.threshold_pct, c.min_delta_ms, c.is_regression,

    c.regression_dimensions == null ? null : JSON.stringify(c.regression_dimensions),

    c.state, c.severity_score, c.updated_at ?? now,

  )),

  ];

  await db.batch(runRewriteStatements);

  return {

    subjects: subjects.length,

    measurements: measurements.length,

    aggregates: aggregates.length,

    comparisons: comparisons.length,

    skipped_no_subject: ingestReport.skipped_no_subject,

  };

}

async function batched(db, statements, chunkSize = 50) {

  for (let index = 0; index < statements.length; index += chunkSize) {

    const chunk = statements.slice(index, index + chunkSize);

    if (chunk.length) {

      await db.batch(chunk);

    }

  }

}

// --- API endpoints (JSON; public pages consume these) -----------------------

async function getScorecard(env) {

  const db = requireDb(env);

  if (!db) {

    return json({ ok: false, error: "d1_not_configured" }, 503);

  }

  const row = await db.prepare(

    "SELECT bench_id, profile, scorecard_json, created_utc FROM scorecards ORDER BY id DESC LIMIT 1",

  ).first();

  if (!row) {

    return json({ ok: false, error: "no_scorecard" }, 404);

  }

  const scorecard = JSON.parse(row.scorecard_json);

  const cells = Array.isArray(scorecard) ? scorecard : scorecard.cells || scorecard.targets || [];

  return json({

    ok: true,

    bench_id: row.bench_id,

    profile: row.profile,

    created_utc: row.created_utc,

    summary: scorecardFromCells(cells),

    scorecard,

  });

}

async function postScorecard(request, env) {

  const db = requireDb(env);

  if (!db) {

    return json({ ok: false, error: "d1_not_configured" }, 503);

  }

  const payload = await readJson(request);

  if (payload instanceof Response) {

    return payload;

  }

  const now = new Date().toISOString();

  await db.prepare(

    "INSERT INTO scorecards (bench_id, profile, scorecard_json, created_utc) VALUES (?, ?, ?, ?)",

  ).bind(

    nullableString(payload.bench_id),

    nullableString(payload.profile),

    JSON.stringify(payload.scorecard ?? payload),

    now,

  ).run();

  return json({ ok: true, created_utc: now });

}

async function getTrends(env, url) {

  const db = requireDb(env);

  if (!db) {

    return json({ ok: false, error: "d1_not_configured" }, 503);

  }

  const scenario = url.searchParams.get("scenario");

  const operation = url.searchParams.get("operation");

  const metric = url.searchParams.get("metric") || "p50";

  if (!scenario || !operation) {

    return json({ ok: false, error: "missing_params", required: ["scenario", "operation"] }, 400);

  }

  const rows = await db.prepare(`

    SELECT r.timestamp_utc, a.run_key, s.name AS subject, a.scenario, a.operation,

           a.p50_ms, a.p90_ms, a.p95_ms, a.p99_ms, a.mean_ms,

           a.avg_tokens, a.avg_output_bytes, a.avg_tool_calls, a.failure_rate

    FROM aggregates a

    JOIN benchmark_runs r ON r.run_key = a.run_key

    JOIN subjects s ON s.subject_id = a.subject_id

    WHERE a.scenario = ? AND a.operation = ?

    ORDER BY r.timestamp_utc ASC

    LIMIT 2000

  `).bind(scenario, operation).all();

  const series = trendSeries(rows.results || [], { scenario, operation, metric });

  if (!series) {

    return json({ ok: false, error: "unknown_metric", metric }, 400);

  }

  return json({ ok: true, ...series });

}

async function getBaselineBook(env, bookId) {

  const db = requireDb(env);

  if (!db) {

    return json({ ok: false, error: "d1_not_configured" }, 503);

  }

  if (!bookId) {

    return json({ ok: false, error: "missing_book_id" }, 400);

  }

  const row = await db.prepare(

    "SELECT book_id, runner_class, manifest_json, book_json, created_utc FROM baseline_books WHERE book_id = ?",

  ).bind(bookId).first();

  if (!row) {

    return json({ ok: false, error: "not_found", book_id: bookId }, 404);

  }

  return json({

    ok: true,

    book_id: row.book_id,

    runner_class: row.runner_class,

    created_utc: row.created_utc,

    manifest: row.manifest_json ? JSON.parse(row.manifest_json) : null,

    book: JSON.parse(row.book_json),

  });

}

async function postBaselineBook(request, env) {

  const db = requireDb(env);

  if (!db) {

    return json({ ok: false, error: "d1_not_configured" }, 503);

  }

  const payload = await readJson(request);

  if (payload instanceof Response) {

    return payload;

  }

  const bookId = stringValue(payload.book_id);

  if (!bookId || typeof payload.book !== "object" || payload.book === null) {

    return json({ ok: false, error: "missing_required_fields", required: ["book_id", "book"] }, 400);

  }

  const now = new Date().toISOString();

  await db.prepare(`

    INSERT INTO baseline_books (book_id, runner_class, manifest_json, book_json, created_utc)

    VALUES (?, ?, ?, ?, ?)

    ON CONFLICT(book_id) DO UPDATE SET

      runner_class = excluded.runner_class,

      manifest_json = excluded.manifest_json,

      book_json = excluded.book_json,

      created_utc = excluded.created_utc

  `).bind(

    bookId,

    nullableString(payload.runner_class),

    payload.manifest === undefined || payload.manifest === null ? null : JSON.stringify(payload.manifest),

    JSON.stringify(payload.book),

    now,

  ).run();

  return json({ ok: true, book_id: bookId, created_utc: now });

}

async function getVerifications(env) {

  const db = requireDb(env);

  if (!db) {

    return json({ ok: false, error: "d1_not_configured" }, 503);

  }

  const rows = await db.prepare(

    "SELECT id, bundle_sha256, verdict, detail_json, created_utc FROM verifications ORDER BY id DESC LIMIT 100",

  ).all();

  return json({

    ok: true,

    verifications: (rows.results || []).map((row) => ({

      id: row.id,

      bundle_sha256: row.bundle_sha256,

      verdict: row.verdict,

      detail: row.detail_json ? JSON.parse(row.detail_json) : null,

      created_utc: row.created_utc,

    })),

  });

}

async function postVerification(request, env) {

  const db = requireDb(env);

  if (!db) {

    return json({ ok: false, error: "d1_not_configured" }, 503);

  }

  const payload = await readJson(request);

  if (payload instanceof Response) {

    return payload;

  }

  const bundleSha = stringValue(payload.bundle_sha256);

  const verdict = stringValue(payload.verdict);

  if (!bundleSha || !verdict) {

    return json({ ok: false, error: "missing_required_fields", required: ["bundle_sha256", "verdict"] }, 400);

  }

  const now = new Date().toISOString();

  await db.prepare(

    "INSERT INTO verifications (bundle_sha256, verdict, detail_json, created_utc) VALUES (?, ?, ?, ?)",

  ).bind(

    bundleSha,

    verdict,

    payload.detail === undefined || payload.detail === null ? null : JSON.stringify(payload.detail),

    now,

  ).run();

  return json({ ok: true, created_utc: now });

}

// --- GitHub webhook (HMAC-verified, recorded into webhook_events) -----------

async function handleGithubWebhook(request, env) {

  if (!env.GITHUB_WEBHOOK_SECRET) {

    return json({ ok: false, error: "webhook_not_configured" }, 503);

  }

  const body = await request.text();

  const valid = await verifySignature(

    env.GITHUB_WEBHOOK_SECRET,

    body,

    request.headers.get("x-hub-signature-256") || "",

  );

  if (!valid) {

    return json({ ok: false, error: "invalid_signature" }, 401);

  }

  const db = requireDb(env);

  if (db) {

    await db.prepare(

      "INSERT INTO webhook_events (event, delivery_id, payload_json, signature_valid, created_utc) VALUES (?, ?, ?, 1, ?)",

    ).bind(

      nullableString(request.headers.get("x-github-event")),

      nullableString(request.headers.get("x-github-delivery")),

      body,

      new Date().toISOString(),

    ).run();

  }

  return json({ ok: true, recorded: Boolean(db) });

}

// --- helpers -----------------------------------------------------------------

function requireDb(env) {

  return env.OAK_BENCH_DB && typeof env.OAK_BENCH_DB.prepare === "function" ? env.OAK_BENCH_DB : null;

}

async function readJson(request) {

  try {

    const payload = await request.json();

    if (!payload || typeof payload !== "object") {

      throw new Error("payload must be a JSON object");

    }

    return payload;

  } catch (error) {

    return json({ ok: false, error: "invalid_json", detail: String(error.message || error) }, 400);

  }

}

function authorize(request, env) {

  const uploadToken = stringValue(env.UPLOAD_TOKEN).trim();

  if (!uploadToken) {

    return json({ ok: false, error: "upload_token_not_configured" }, 503);

  }

  const expected = `Bearer ${uploadToken}`;

  if (request.headers.get("authorization") === expected) {

    return null;

  }

  return json({ ok: false, error: "unauthorized" }, 401);

}

function parseJsonl(text) {

  const lines = text.split(/\r?\n/);

  const rows = [];

  for (let index = 0; index < lines.length; index += 1) {

    const line = lines[index].trim();

    if (!line) {

      continue;

    }

    const row = JSON.parse(line);

    if (!row || typeof row !== "object" || Array.isArray(row)) {

      throw new Error(`line ${index + 1} is not a JSON object`);

    }

    rows.push(row);

  }

  if (rows.length === 0) {

    throw new Error("upload contained no JSON rows");

  }

  return { firstRow: rows[0], rows, rowCount: rows.length };

}

async function persistUploadMetadata(db, upload) {

  const row = upload.firstRow;

  const source = row.source && typeof row.source === "object" ? row.source : {};

  const subjectVersions = row.subject_versions && typeof row.subject_versions === "object"

    ? row.subject_versions

    : {};

  const subjectDetails = row.subject_details && typeof row.subject_details === "object"

    ? row.subject_details

    : {};

  const hostId = await stableId([

    stringValue(row.host),

    stringValue(row.platform),

    stringValue(row.machine),

    stringValue(row.python),

  ].join("\u001f"));

  await db.batch([

    db.prepare(`

      INSERT INTO hosts (

        host_id, host, platform, machine, python, metadata_json, first_seen_at, last_seen_at

      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)

      ON CONFLICT(host_id) DO UPDATE SET

        last_seen_at = excluded.last_seen_at,

        metadata_json = excluded.metadata_json

    `).bind(

      hostId,

      nullableString(row.host),

      nullableString(row.platform),

      nullableString(row.machine),

      nullableString(row.python),

      JSON.stringify({ source: "worker-upload" }),

      upload.now,

      upload.now,

    ),

    db.prepare(`

      INSERT INTO uploads (

        r2_key, sha256, bench_id, profile, row_count, byte_count, content_type, status, uploaded_at

      ) VALUES (?, ?, ?, ?, ?, ?, ?, 'accepted', ?)

      ON CONFLICT(r2_key) DO UPDATE SET

        row_count = excluded.row_count,

        byte_count = excluded.byte_count,

        status = excluded.status,

        uploaded_at = excluded.uploaded_at

    `).bind(

      upload.r2Key,

      upload.sha256,

      stringValue(row.bench_id),

      stringValue(row.profile),

      upload.rowCount,

      upload.byteCount,

      upload.contentType,

      upload.now,

    ),

    db.prepare(`

      INSERT INTO benchmark_runs (

        run_key, bench_id, timestamp_utc, profile, source_oak_hash, source_oak_repo,

        source_oak_status, host_id, raw_r2_key, benchmark_track, command_semantics_version,

        row_count, subject_versions_json, subject_details_json, uploaded_at

      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

      ON CONFLICT(run_key) DO UPDATE SET

        raw_r2_key = excluded.raw_r2_key,

        benchmark_track = excluded.benchmark_track,

        command_semantics_version = excluded.command_semantics_version,

        row_count = excluded.row_count,

        subject_versions_json = excluded.subject_versions_json,

        subject_details_json = excluded.subject_details_json,

        uploaded_at = excluded.uploaded_at

    `).bind(

      `${stringValue(row.bench_id)}:${stringValue(row.profile)}`,

      stringValue(row.bench_id),

      stringValue(row.timestamp_utc || row.bench_id),

      stringValue(row.profile),

      nullableString(source.oak_hash),

      nullableString(source.oak_repo),

      nullableString(source.oak_status),

      hostId,

      upload.r2Key,

      nullableString(row.benchmark_track),

      nullableString(row.command_semantics_version),

      upload.rowCount,

      JSON.stringify(subjectVersions),

      JSON.stringify(subjectDetails),

      upload.now,

    ),

  ]);

}

function objectKey(profile, timestampUtc, benchId, sha256) {

  const date = isoDate(timestampUtc);

  return [

    "raw/v1",

    `profile=${sanitizePathPart(profile)}`,

    `date=${date}`,

    `bench_id=${sanitizePathPart(benchId)}`,

    `sha256=${sha256.slice(0, 12)}`,

    "results.jsonl",

  ].join("/");

}

function isoDate(timestampUtc) {

  const compact = String(timestampUtc || "");

  const match = compact.match(/^(\d{4})(\d{2})(\d{2})T/);

  if (match) {

    return `${match[1]}-${match[2]}-${match[3]}`;

  }

  const parsed = new Date(compact);

  if (!Number.isNaN(parsed.getTime())) {

    return parsed.toISOString().slice(0, 10);

  }

  return new Date().toISOString().slice(0, 10);

}

function sanitizePathPart(value) {

  return String(value || "unknown").replace(/[^A-Za-z0-9._=-]/g, "_");

}

function stringValue(value) {

  if (value === null || value === undefined) {

    return "";

  }

  return String(value);

}

function nullableString(value) {

  const text = stringValue(value);

  return text ? text : null;

}

function json(payload, status = 200) {

  return new Response(JSON.stringify(payload, null, 2), {

    status,

    headers: { "content-type": "application/json; charset=utf-8" },

  });

}