Build an e-commerce catalog assistant with safe recommendations

{
  "name": "bounded-catalog-assistant",
  "private": true,
  "type": "module",
  "engines": {
    "node": "22.22.2"
  },
  "scripts": {
    "start": "node --import ./src/telemetry.mjs src/server.mjs",
    "sync": "node src/sync-catalog.mjs",
    "check": "node --check src/policy.mjs && node --check src/server.mjs && node --check src/sync-catalog.mjs && node --check src/telemetry.mjs",
    "test": "node --test test/catalog-policy.test.mjs"
  },
  "dependencies": {
    "@opentelemetry/auto-instrumentations-node": "0.67.2",
    "@opentelemetry/exporter-metrics-otlp-http": "0.208.0",
    "@opentelemetry/exporter-trace-otlp-http": "0.208.0",
    "@opentelemetry/sdk-node": "0.208.0",
    "openai": "7.1.0",
    "pg": "8.16.3",
    "zod": "4.4.3"
  }
}

PORT=8090
DATABASE_URL=postgresql://catalog_app:replace-me@postgres:5432/catalog
OPENAI_API_KEY=replace-in-secret-manager
OPENAI_MODEL=gpt-5.4-nano
OPENAI_EMBEDDING_MODEL=text-embedding-3-small
CLIENT_BEARER_TOKEN=replace-with-32-random-bytes
MERCHANT_OPERATOR_TOKEN=replace-with-32-random-bytes
HANDOFF_DISPATCH_TOKEN=replace-with-32-random-bytes
SUPPORT_WEBHOOK_URL=https://support.example.invalid/v1/handoffs
SUPPORT_WEBHOOK_HOST=support.example.invalid
SUPPORT_WEBHOOK_TOKEN=replace-with-destination-specific-token
CATALOG_URL_HOSTS=shop.example.invalid
INVENTORY_MAX_AGE_SECONDS=120
REQUESTS_PER_MINUTE=60
MAX_QUERY_BYTES=16384
OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318

BEGIN;
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE IF NOT EXISTS catalog_revisions (
  revision text PRIMARY KEY,
  source_hash text NOT NULL,
  imported_by text NOT NULL,
  imported_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS products (
  id uuid PRIMARY KEY,
  sku text UNIQUE NOT NULL CHECK (char_length(sku) BETWEEN 1 AND 80),
  title text NOT NULL CHECK (char_length(title) BETWEEN 2 AND 240),
  description text NOT NULL CHECK (octet_length(description) <= 16000),
  category text NOT NULL CHECK (char_length(category) BETWEEN 1 AND 120),
  attributes jsonb NOT NULL CHECK (jsonb_typeof(attributes) = 'object'),
  price_minor integer NOT NULL CHECK (price_minor >= 0),
  currency text NOT NULL CHECK (currency ~ '^[A-Z]{3}$'),
  canonical_url text NOT NULL,
  active boolean NOT NULL DEFAULT true,
  catalog_revision text NOT NULL REFERENCES catalog_revisions(revision),
  updated_at timestamptz NOT NULL
);

CREATE TABLE IF NOT EXISTS inventory_snapshots (
  product_id uuid PRIMARY KEY REFERENCES products(id) ON DELETE CASCADE,
  available boolean NOT NULL,
  quantity integer CHECK (quantity IS NULL OR quantity >= 0),
  observed_at timestamptz NOT NULL,
  source_revision text NOT NULL
);

CREATE TABLE IF NOT EXISTS product_embeddings (
  product_id uuid PRIMARY KEY REFERENCES products(id) ON DELETE CASCADE,
  model text NOT NULL,
  content_hash text NOT NULL,
  embedding vector(1536) NOT NULL,
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS handoff_drafts (
  id uuid PRIMARY KEY,
  session_id uuid NOT NULL,
  reason text NOT NULL CHECK (octet_length(reason) <= 1200),
  transcript_summary text NOT NULL CHECK (octet_length(transcript_summary) <= 4000),
  product_ids uuid[] NOT NULL DEFAULT '{}',
  status text NOT NULL CHECK (status IN ('draft','approved','dispatched','rejected')),
  approved_by text,
  approved_at timestamptz,
  dispatch_id text,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS commerce_audit (
  sequence bigserial PRIMARY KEY,
  session_id uuid,
  actor_hash text NOT NULL,
  event_type text NOT NULL,
  object_id text,
  detail jsonb NOT NULL DEFAULT '{}'::jsonb,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS commerce_rate_buckets (
  key text NOT NULL,
  bucket timestamptz NOT NULL,
  count integer NOT NULL CHECK (count > 0),
  PRIMARY KEY (key, bucket)
);

CREATE INDEX IF NOT EXISTS product_category_active
  ON products (category, active);
CREATE INDEX IF NOT EXISTS product_embedding_hnsw
  ON product_embeddings USING hnsw (embedding vector_cosine_ops);
CREATE INDEX IF NOT EXISTS handoff_dispatch_queue
  ON handoff_drafts (status, created_at) WHERE status = 'approved';
COMMIT;

import { createHash, timingSafeEqual } from "node:crypto";
import { z } from "zod";

export const ProductImport = z.object({
  id: z.string().uuid(),
  sku: z.string().min(1).max(80),
  title: z.string().min(2).max(240),
  description: z.string().min(1).max(12000),
  category: z.string().min(1).max(120),
  attributes: z.record(z.string(), z.union([z.string(), z.number(), z.boolean()])),
  price_minor: z.number().int().nonnegative(),
  currency: z.string().regex(/^[A-Z]{3}$/),
  canonical_url: z.string().url().refine((value) => new URL(value).protocol === "https:"),
  active: z.boolean(),
  updated_at: z.string().datetime(),
  inventory: z.object({
    available: z.boolean(),
    quantity: z.number().int().nonnegative().nullable(),
    observed_at: z.string().datetime(),
    source_revision: z.string().min(1).max(120)
  })
}).strict();

export const QueryPlan = z.object({
  semantic_query: z.string().min(2).max(500),
  category: z.string().min(1).max(120).nullable(),
  currency: z.string().regex(/^[A-Z]{3}$/).nullable(),
  min_price_minor: z.number().int().nonnegative().nullable(),
  max_price_minor: z.number().int().nonnegative().nullable(),
  must_have: z.array(z.string().min(1).max(80)).max(10),
  max_results: z.number().int().min(1).max(8),
  wants_human: z.boolean()
}).strict().superRefine((value, context) => {
  if (
    value.min_price_minor !== null &&
    value.max_price_minor !== null &&
    value.min_price_minor > value.max_price_minor
  ) {
    context.addIssue({ code: "custom", message: "price_range_invalid" });
  }
  if (
    (value.min_price_minor !== null || value.max_price_minor !== null) &&
    value.currency === null
  ) {
    context.addIssue({ code: "custom", message: "price_currency_required" });
  }
});

export const CatalogAnswer = z.object({
  summary: z.string().min(20).max(1200),
  product_ids: z.array(z.string().uuid()).max(5),
  reasons: z.array(z.object({
    product_id: z.string().uuid(),
    reason: z.string().min(10).max(500)
  })).max(5),
  follow_up_question: z.string().min(5).max(300).nullable(),
  handoff: z.object({
    requested: z.boolean(),
    reason: z.string().max(800),
    transcript_summary: z.string().max(2000)
  })
}).strict();

const forbiddenClaims =
  /(?:\$\s*\d|€\s*\d|£\s*\d|\b(?:in stock|out of stock|only \d+ left|guaranteed|best for everyone|will cure|will treat)\b)/i;
const sensitivePreference =
  /\b(?:race|ethnicity|religion|sexual orientation|disability|medical diagnosis|credit score)\b/i;

export function contentForEmbedding(product) {
  const attributes = Object.entries(product.attributes)
    .sort(([left], [right]) => left.localeCompare(right))
    .map(([key, value]) => key + ": " + String(value))
    .join("; ");
  return [
    product.title,
    product.category,
    product.description,
    attributes
  ].join("\n").slice(0, 16000);
}

export function contentHash(product) {
  return createHash("sha256").update(contentForEmbedding(product)).digest("hex");
}

export function vectorLiteral(values) {
  if (
    !Array.isArray(values) ||
    values.length !== 1536 ||
    values.some((value) => !Number.isFinite(value))
  ) throw new Error("embedding_shape_invalid");
  return "[" + values.join(",") + "]";
}

export function stockState(inventory, now, maximumAgeSeconds) {
  if (!inventory) return { status: "unknown", reason: "missing_snapshot" };
  const ageSeconds = Math.floor(
    (now.valueOf() - new Date(inventory.observed_at).valueOf()) / 1000
  );
  if (!Number.isFinite(ageSeconds) || ageSeconds < 0 || ageSeconds > maximumAgeSeconds) {
    return { status: "unknown", reason: "stale_snapshot", age_seconds: ageSeconds };
  }
  if (!inventory.available || inventory.quantity === 0) {
    return { status: "out_of_stock", observed_at: inventory.observed_at, age_seconds: ageSeconds };
  }
  return {
    status: "available",
    quantity: inventory.quantity,
    observed_at: inventory.observed_at,
    age_seconds: ageSeconds
  };
}

export function validateAnswer(raw, candidates, query) {
  if (sensitivePreference.test(query)) throw new Error("sensitive_preference_not_supported");
  const answer = CatalogAnswer.parse(raw);
  const allowed = new Set(candidates.map((item) => item.id));
  const referenced = [...answer.product_ids, ...answer.reasons.map((item) => item.product_id)];
  if (referenced.some((id) => !allowed.has(id))) throw new Error("unknown_product_reference");
  const unique = new Set(answer.product_ids);
  if (unique.size !== answer.product_ids.length) throw new Error("duplicate_product_reference");
  if (answer.reasons.some((item) => !unique.has(item.product_id))) {
    throw new Error("reason_without_selected_product");
  }
  const modelText = [
    answer.summary,
    ...answer.reasons.map((item) => item.reason),
    answer.follow_up_question || ""
  ].join(" ");
  if (forbiddenClaims.test(modelText)) throw new Error("catalog_fact_claim_from_model");
  return answer;
}

export function publicProduct(product, now, maximumAgeSeconds) {
  return {
    id: product.id,
    sku: product.sku,
    title: product.title,
    category: product.category,
    attributes: product.attributes,
    price: {
      amount_minor: product.price_minor,
      currency: product.currency
    },
    url: product.canonical_url,
    stock: stockState(product, now, maximumAgeSeconds)
  };
}

export function safeEqualBearer(supplied, expected) {
  const left = Buffer.from(String(supplied || ""));
  const right = Buffer.from("Bearer " + expected);
  return left.length === right.length && timingSafeEqual(left, right);
}

import fs from "node:fs/promises";
import { createHash } from "node:crypto";
import OpenAI from "openai";
import pg from "pg";
import {
  contentForEmbedding,
  contentHash,
  ProductImport,
  vectorLiteral
} from "./policy.mjs";

const required = (name) => {
  const value = process.env[name];
  if (!value) throw new Error("missing_" + name.toLowerCase());
  return value;
};

const inputPath = process.argv[2];
const revision = process.argv[3];
if (!inputPath || !revision) {
  throw new Error("usage: npm run sync -- <catalog.ndjson> <revision>");
}

const raw = await fs.readFile(inputPath, "utf8");
if (Buffer.byteLength(raw) > 20 * 1024 * 1024) throw new Error("catalog_file_too_large");
const products = raw.split(/\r?\n/).filter(Boolean).map((line) => ProductImport.parse(JSON.parse(line)));
if (!products.length || products.length > 20_000) throw new Error("catalog_count_out_of_bounds");
if (new Set(products.map((item) => item.id)).size !== products.length) {
  throw new Error("duplicate_product_id");
}
const allowedCatalogHosts = new Set(
  required("CATALOG_URL_HOSTS").split(",").map((item) => item.trim().toLowerCase())
);
if (products.some((item) => !allowedCatalogHosts.has(new URL(item.canonical_url).hostname.toLowerCase()))) {
  throw new Error("catalog_url_host_not_allowed");
}

const sourceHash = createHash("sha256").update(raw).digest("hex");
const pool = new pg.Pool({ connectionString: required("DATABASE_URL"), max: 4 });
const openai = new OpenAI({ apiKey: required("OPENAI_API_KEY"), timeout: 30_000, maxRetries: 2 });
const model = required("OPENAI_EMBEDDING_MODEL");
const actorHash = createHash("sha256")
  .update(required("MERCHANT_OPERATOR_TOKEN"))
  .digest("hex");
const client = await pool.connect();

try {
  await client.query("BEGIN");
  await client.query(
    "INSERT INTO catalog_revisions(revision,source_hash,imported_by) VALUES($1,$2,$3)",
    [revision, sourceHash, actorHash]
  );
  let embedded = 0;
  let unchanged = 0;
  for (const product of products) {
    const hash = contentHash(product);
    const previous = await client.query(
      "SELECT content_hash FROM product_embeddings WHERE product_id=$1",
      [product.id]
    );
    let vector = null;
    if (!previous.rowCount || previous.rows[0].content_hash !== hash) {
      const response = await openai.embeddings.create({
        model,
        input: contentForEmbedding(product),
        encoding_format: "float"
      });
      vector = vectorLiteral(response.data[0].embedding);
      embedded += 1;
    } else {
      unchanged += 1;
    }
    await client.query(
      "INSERT INTO products(id,sku,title,description,category,attributes,price_minor,currency,canonical_url,active,catalog_revision,updated_at) " +
        "VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) " +
        "ON CONFLICT(id) DO UPDATE SET sku=EXCLUDED.sku,title=EXCLUDED.title,description=EXCLUDED.description," +
        "category=EXCLUDED.category,attributes=EXCLUDED.attributes,price_minor=EXCLUDED.price_minor," +
        "currency=EXCLUDED.currency,canonical_url=EXCLUDED.canonical_url,active=EXCLUDED.active," +
        "catalog_revision=EXCLUDED.catalog_revision,updated_at=EXCLUDED.updated_at",
      [
        product.id, product.sku, product.title, product.description, product.category,
        product.attributes, product.price_minor, product.currency, product.canonical_url,
        product.active, revision, product.updated_at
      ]
    );
    await client.query(
      "INSERT INTO inventory_snapshots(product_id,available,quantity,observed_at,source_revision) " +
        "VALUES($1,$2,$3,$4,$5) ON CONFLICT(product_id) DO UPDATE SET " +
        "available=EXCLUDED.available,quantity=EXCLUDED.quantity,observed_at=EXCLUDED.observed_at," +
        "source_revision=EXCLUDED.source_revision",
      [
        product.id,
        product.inventory.available,
        product.inventory.quantity,
        product.inventory.observed_at,
        product.inventory.source_revision
      ]
    );
    if (vector) {
      await client.query(
        "INSERT INTO product_embeddings(product_id,model,content_hash,embedding) VALUES($1,$2,$3,$4::vector) " +
          "ON CONFLICT(product_id) DO UPDATE SET model=EXCLUDED.model,content_hash=EXCLUDED.content_hash," +
          "embedding=EXCLUDED.embedding,updated_at=now()",
        [product.id, model, hash, vector]
      );
    }
  }
  await client.query("COMMIT");
  console.log(JSON.stringify({
    status: "CATALOG_SYNC_OK",
    revision,
    products: products.length,
    embedded,
    unchanged,
    source_hash: sourceHash
  }));
} catch (error) {
  await client.query("ROLLBACK");
  throw error;
} finally {
  client.release();
  await pool.end();
}

import http from "node:http";
import { createHash, randomUUID } from "node:crypto";
import OpenAI from "openai";
import pg from "pg";
import {
  CatalogAnswer,
  publicProduct,
  QueryPlan,
  safeEqualBearer,
  validateAnswer,
  vectorLiteral
} from "./policy.mjs";

const required = (name) => {
  const value = process.env[name];
  if (!value) throw new Error("missing_" + name.toLowerCase());
  return value;
};

const pool = new pg.Pool({ connectionString: required("DATABASE_URL"), max: 12 });
const openai = new OpenAI({ apiKey: required("OPENAI_API_KEY"), timeout: 30_000, maxRetries: 1 });
const model = required("OPENAI_MODEL");
const embeddingModel = required("OPENAI_EMBEDDING_MODEL");
const maxAge = Number(process.env.INVENTORY_MAX_AGE_SECONDS || 120);
const perMinute = Number(process.env.REQUESTS_PER_MINUTE || 60);
const maxQueryBytes = Number(process.env.MAX_QUERY_BYTES || 16384);

function fixedWebhook() {
  const value = new URL(required("SUPPORT_WEBHOOK_URL"));
  if (
    value.protocol !== "https:" ||
    value.username ||
    value.password ||
    value.hostname !== required("SUPPORT_WEBHOOK_HOST")
  ) throw new Error("support_webhook_configuration_invalid");
  return value.href;
}

function json(response, status, body) {
  const output = JSON.stringify(body);
  response.writeHead(status, {
    "content-type": "application/json",
    "content-length": Buffer.byteLength(output),
    "cache-control": "no-store"
  });
  response.end(output);
}

async function readJson(request) {
  const chunks = [];
  let size = 0;
  for await (const chunk of request) {
    size += chunk.length;
    if (size > maxQueryBytes) throw Object.assign(new Error("payload_too_large"), { status: 413 });
    chunks.push(chunk);
  }
  try {
    return JSON.parse(Buffer.concat(chunks).toString("utf8"));
  } catch {
    throw Object.assign(new Error("invalid_json"), { status: 400 });
  }
}

function authenticate(request, secretName, role) {
  if (!safeEqualBearer(request.headers.authorization, required(secretName))) {
    throw Object.assign(new Error(role + "_authentication_failed"), { status: 401 });
  }
  return createHash("sha256")
    .update(role + ":" + request.headers.authorization)
    .digest("hex");
}

async function rateLimit(key) {
  const result = await pool.query(
    "INSERT INTO commerce_rate_buckets(key,bucket,count) VALUES($1,date_trunc('minute',now()),1) " +
      "ON CONFLICT(key,bucket) DO UPDATE SET count=commerce_rate_buckets.count+1 " +
      "WHERE commerce_rate_buckets.count < $2 RETURNING count",
    [key, perMinute]
  );
  if (!result.rowCount) throw Object.assign(new Error("rate_limit_exceeded"), { status: 429 });
}

async function structured(input, schema, name, instructions, maxOutputTokens) {
  const result = await openai.responses.create({
    model,
    store: false,
    instructions,
    input,
    max_output_tokens: maxOutputTokens,
    text: {
      format: {
        type: "json_schema",
        name,
        strict: true,
        schema: schema.toJSONSchema()
      }
    }
  });
  if (result.status !== "completed" || !result.output_text) {
    throw Object.assign(new Error("model_response_incomplete"), { status: 502 });
  }
  return { id: result.id, value: JSON.parse(result.output_text), usage: result.usage };
}

async function searchCatalog(query, plan) {
  const embedded = await openai.embeddings.create({
    model: embeddingModel,
    input: plan.semantic_query,
    encoding_format: "float"
  });
  const vector = vectorLiteral(embedded.data[0].embedding);
  const result = await pool.query(
    "SELECT p.*, i.available, i.quantity, i.observed_at, i.source_revision, " +
      "1 - (e.embedding <=> $1::vector) AS semantic_score " +
      "FROM product_embeddings e JOIN products p ON p.id=e.product_id " +
      "LEFT JOIN inventory_snapshots i ON i.product_id=p.id " +
      "WHERE p.active=true AND ($2::text IS NULL OR p.category=$2) " +
      "AND ($3::text IS NULL OR p.currency=$3) " +
      "AND ($4::integer IS NULL OR p.price_minor >= $4) " +
      "AND ($5::integer IS NULL OR p.price_minor <= $5) " +
      "AND (cardinality($6::text[]) = 0 OR p.attributes ?& $6::text[]) " +
      "ORDER BY e.embedding <=> $1::vector, p.id LIMIT $7",
    [
      vector,
      plan.category,
      plan.currency,
      plan.min_price_minor,
      plan.max_price_minor,
      plan.must_have,
      plan.max_results
    ]
  );
  return result.rows;
}

async function createHandoff(client, sessionId, actorHash, answer) {
  if (!answer.handoff.requested) return null;
  const id = randomUUID();
  await client.query(
    "INSERT INTO handoff_drafts(id,session_id,reason,transcript_summary,product_ids,status) " +
      "VALUES($1,$2,$3,$4,$5,'draft')",
    [
      id,
      sessionId,
      answer.handoff.reason,
      answer.handoff.transcript_summary,
      answer.product_ids
    ]
  );
  await client.query(
    "INSERT INTO commerce_audit(session_id,actor_hash,event_type,object_id) VALUES($1,$2,'handoff_drafted',$3)",
    [sessionId, actorHash, id]
  );
  return id;
}

async function answerSearch(request, response) {
  const actorHash = authenticate(request, "CLIENT_BEARER_TOKEN", "client");
  await rateLimit("search:" + actorHash);
  const input = await readJson(request);
  if (typeof input.query !== "string" || input.query.length < 2 || input.query.length > 2000) {
    throw Object.assign(new Error("query_invalid"), { status: 400 });
  }
  const sessionId = input.session_id ? String(input.session_id) : randomUUID();
  const planned = await structured(
    input.query,
    QueryPlan,
    "catalog_query_plan",
    "Extract only explicit catalog constraints. Use ISO currency when a price is present. Put required structured attribute keys in must_have. Do not infer sensitive traits. Set wants_human for order, account, complaint, regulated, or uncertain requests.",
    500
  );
  const plan = QueryPlan.parse(planned.value);
  const candidates = await searchCatalog(input.query, plan);
  const candidateView = candidates.map((item) => ({
    id: item.id,
    sku: item.sku,
    title: item.title,
    description: item.description,
    category: item.category,
    attributes: item.attributes,
    semantic_score: Number(item.semantic_score)
  }));
  const generated = await structured(
    JSON.stringify({ query: input.query, plan, candidates: candidateView }),
    CatalogAnswer,
    "catalog_answer",
    "Recommend only supplied product IDs. Explain attribute fit only. Never state price, stock, scarcity, guarantee, universal best, medical benefit, or order status. Request handoff for account, order, complaint, regulated, or unresolved needs.",
    1200
  );
  const answer = validateAnswer(generated.value, candidates, input.query);
  if (plan.wants_human && !answer.handoff.requested) {
    answer.handoff = {
      requested: true,
      reason: "The query requires a human-owned account, order, complaint, regulated, or unresolved workflow.",
      transcript_summary: "Customer requested human assistance after a bounded catalog search."
    };
  }
  const client = await pool.connect();
  let handoffId;
  try {
    await client.query("BEGIN");
    handoffId = await createHandoff(client, sessionId, actorHash, answer);
    await client.query(
      "INSERT INTO commerce_audit(session_id,actor_hash,event_type,object_id,detail) " +
        "VALUES($1,$2,'catalog_answered',$3,$4)",
      [
        sessionId,
        actorHash,
        generated.id,
        {
          plan_response_id: planned.id,
          candidate_ids: candidates.map((item) => item.id),
          selected_ids: answer.product_ids,
          handoff_id: handoffId,
          model
        }
      ]
    );
    await client.query("COMMIT");
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  } finally {
    client.release();
  }
  const byId = new Map(candidates.map((item) => [item.id, item]));
  json(response, 200, {
    session_id: sessionId,
    summary: answer.summary,
    recommendations: answer.product_ids.map((id) => ({
      product: publicProduct(byId.get(id), new Date(), maxAge),
      reason: answer.reasons.find((item) => item.product_id === id)?.reason || null
    })),
    follow_up_question: answer.follow_up_question,
    handoff: handoffId ? { id: handoffId, status: "draft", dispatched: false } : null,
    catalog_facts_source: "database",
    external_writes: 0
  });
}

async function approveHandoff(request, response, id) {
  const actorHash = authenticate(request, "MERCHANT_OPERATOR_TOKEN", "merchant_operator");
  const result = await pool.query(
    "UPDATE handoff_drafts SET status='approved',approved_by=$1,approved_at=now() " +
      "WHERE id=$2 AND status='draft' RETURNING session_id,id",
    [actorHash, id]
  );
  if (!result.rowCount) throw Object.assign(new Error("handoff_not_approvable"), { status: 409 });
  await pool.query(
    "INSERT INTO commerce_audit(session_id,actor_hash,event_type,object_id) " +
      "VALUES($1,$2,'handoff_approved',$3)",
    [result.rows[0].session_id, actorHash, id]
  );
  json(response, 200, { handoff_id: id, status: "approved", dispatched: false });
}

async function dispatchHandoff(request, response, id) {
  const actorHash = authenticate(request, "HANDOFF_DISPATCH_TOKEN", "handoff_dispatcher");
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const selected = await client.query(
      "SELECT * FROM handoff_drafts WHERE id=$1 AND status='approved' FOR UPDATE",
      [id]
    );
    if (!selected.rowCount) throw Object.assign(new Error("handoff_not_dispatchable"), { status: 409 });
    const draft = selected.rows[0];
    const sent = await fetch(fixedWebhook(), {
      method: "POST",
      redirect: "error",
      headers: {
        "authorization": "Bearer " + required("SUPPORT_WEBHOOK_TOKEN"),
        "content-type": "application/json",
        "idempotency-key": draft.id
      },
      body: JSON.stringify({
        handoff_id: draft.id,
        session_id: draft.session_id,
        reason: draft.reason,
        transcript_summary: draft.transcript_summary,
        product_ids: draft.product_ids
      }),
      signal: AbortSignal.timeout(10_000)
    });
    if (!sent.ok) throw Object.assign(new Error("support_dispatch_failed"), { status: 502 });
    const dispatchId = sent.headers.get("x-request-id") || draft.id;
    await client.query(
      "UPDATE handoff_drafts SET status='dispatched',dispatch_id=$1 WHERE id=$2",
      [dispatchId, draft.id]
    );
    await client.query(
      "INSERT INTO commerce_audit(session_id,actor_hash,event_type,object_id,detail) " +
        "VALUES($1,$2,'handoff_dispatched',$3,$4)",
      [draft.session_id, actorHash, draft.id, { dispatch_id: dispatchId }]
    );
    await client.query("COMMIT");
    json(response, 200, { handoff_id: draft.id, status: "dispatched", dispatch_id: dispatchId });
  } catch (error) {
    await client.query("ROLLBACK");
    throw error;
  } finally {
    client.release();
  }
}

const server = http.createServer(async (request, response) => {
  const requestId = randomUUID();
  const started = performance.now();
  try {
    const url = new URL(request.url, "http://localhost");
    const parts = url.pathname.split("/").filter(Boolean);
    if (request.method === "GET" && url.pathname === "/healthz") {
      const database = await pool.query(
        "SELECT max(imported_at) AS catalog_at, " +
          "(SELECT max(observed_at) FROM inventory_snapshots) AS inventory_at FROM catalog_revisions"
      );
      return json(response, 200, { status: "ok", ...database.rows[0], request_id: requestId });
    }
    if (request.method === "POST" && url.pathname === "/v1/search") {
      return await answerSearch(request, response);
    }
    if (request.method === "POST" && parts[0] === "v1" && parts[1] === "handoffs" && parts[3] === "approve") {
      return await approveHandoff(request, response, parts[2]);
    }
    if (request.method === "POST" && parts[0] === "v1" && parts[1] === "handoffs" && parts[3] === "dispatch") {
      return await dispatchHandoff(request, response, parts[2]);
    }
    json(response, 404, { error: "not_found", request_id: requestId });
  } catch (error) {
    const status = Number(error.status) || 500;
    console.error(JSON.stringify({
      level: "error",
      request_id: requestId,
      code: status >= 500 ? "internal_error" : error.message,
      duration_ms: Math.round(performance.now() - started)
    }));
    json(response, status, {
      error: status >= 500 ? "internal_error" : error.message,
      request_id: requestId
    });
  }
});

server.listen(Number(process.env.PORT || 8090), "0.0.0.0", () => {
  console.log(JSON.stringify({ level: "info", event: "catalog_assistant_started", port: Number(process.env.PORT || 8090) }));
});

import assert from "node:assert/strict";
import test from "node:test";
import {
  contentHash,
  ProductImport,
  publicProduct,
  stockState,
  validateAnswer,
  vectorLiteral
} from "../src/policy.mjs";

const product = ProductImport.parse({
  id: "1d334b9d-cad5-4385-a3d5-430cacb238ee",
  sku: "PACK-18-BLUE",
  title: "18 litre blue daypack",
  description: "A compact recycled-nylon daypack with padded laptop sleeve.",
  category: "bags",
  attributes: { color: "blue", capacity_litres: 18, laptop_sleeve: true },
  price_minor: 7900,
  currency: "USD",
  canonical_url: "https://shop.example.invalid/products/pack-18-blue",
  active: true,
  updated_at: "2026-07-29T12:00:00.000Z",
  inventory: {
    available: true,
    quantity: 4,
    observed_at: "2026-07-29T12:00:00.000Z",
    source_revision: "inventory-884"
  }
});

test("structured product import and content hash are deterministic", () => {
  assert.equal(product.sku, "PACK-18-BLUE");
  assert.equal(contentHash(product).length, 64);
  assert.equal(contentHash(product), contentHash({
    ...product,
    attributes: { laptop_sleeve: true, capacity_litres: 18, color: "blue" }
  }));
});

test("inventory freshness is fail-closed", () => {
  assert.equal(
    stockState(product.inventory, new Date("2026-07-29T12:01:00.000Z"), 120).status,
    "available"
  );
  assert.equal(
    stockState(product.inventory, new Date("2026-07-29T12:05:00.000Z"), 120).status,
    "unknown"
  );
  assert.equal(stockState(null, new Date(), 120).status, "unknown");
});

test("model can reference candidates but cannot author catalog facts", () => {
  const answer = {
    summary: "This daypack matches the requested compact size and laptop sleeve.",
    product_ids: [product.id],
    reasons: [{
      product_id: product.id,
      reason: "Its structured attributes include an 18 litre capacity and laptop sleeve."
    }],
    follow_up_question: "Would another color be useful?",
    handoff: { requested: false, reason: "", transcript_summary: "" }
  };
  assert.equal(validateAnswer(answer, [product], "small laptop bag").product_ids[0], product.id);
  assert.throws(
    () => validateAnswer({ ...answer, summary: "It is in stock for $79." }, [product], "small laptop bag"),
    /catalog_fact_claim/
  );
  assert.throws(
    () => validateAnswer({
      ...answer,
      product_ids: ["1d334b9d-cad5-4385-a3d5-430cacb23999"],
      reasons: []
    }, [product], "small laptop bag"),
    /unknown_product/
  );
});

test("public product renders price and stock from database state", () => {
  const output = publicProduct({
    ...product,
    ...product.inventory
  }, new Date("2026-07-29T12:01:00.000Z"), 120);
  assert.deepEqual(output.price, { amount_minor: 7900, currency: "USD" });
  assert.equal(output.stock.status, "available");
  assert.equal(output.stock.quantity, 4);
});

test("embedding vector must match reviewed model dimensions", () => {
  assert.match(vectorLiteral(Array(1536).fill(0)), /^\[0,0,0/);
  assert.throws(() => vectorLiteral([0, 1]), /embedding_shape_invalid/);
});

import { NodeSDK } from "@opentelemetry/sdk-node";
import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";

const endpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
if (endpoint) {
  const sdk = new NodeSDK({
    serviceName: "catalog-assistant",
    traceExporter: new OTLPTraceExporter({ url: endpoint + "/v1/traces" }),
    metricReader: new PeriodicExportingMetricReader({
      exporter: new OTLPMetricExporter({ url: endpoint + "/v1/metrics" }),
      exportIntervalMillis: 30_000
    }),
    instrumentations: [getNodeAutoInstrumentations()]
  });
  await sdk.start();
  process.on("SIGTERM", () => sdk.shutdown().finally(() => process.exit(0)));
}

services:
  postgres:
    image: pgvector/pgvector:pg17@sha256:replace-with-reviewed-digest
    environment:
      POSTGRES_DB: catalog
      POSTGRES_USER: catalog_app
      POSTGRES_PASSWORD_FILE: /run/secrets/postgres_password
    secrets: [postgres_password]
    volumes: [catalog-db:/var/lib/postgresql/data]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U catalog_app -d catalog"]
      interval: 10s
      timeout: 3s
      retries: 10
  app:
    build: .
    read_only: true
    tmpfs: ["/tmp:rw,noexec,nosuid,size=32m"]
    env_file: .env
    depends_on:
      postgres:
        condition: service_healthy
    ports: ["127.0.0.1:8090:8090"]
    restart: unless-stopped
volumes:
  catalog-db:
secrets:
  postgres_password:
    file: ./secrets/postgres_password