feat: PGVector-backed log indexing and semantic search (#24)

app/api/inngest/route.ts

@@ -6,11 +6,12 @@ import { runApiTests } from "@/lib/test-functions/api-tests";
 import { runPerformanceTests } from "@/lib/test-functions/performance-tests";
 import { runVibetest } from "@/lib/test-functions/vibetest-run";
 import { runSecurityAgent } from "@/lib/security-agent/functions";
+import { indexDeploymentLogs, cronReindexLogs } from "@/lib/vector-indexer"; // ← add this
 
 // Allow Inngest handler up to 5 minutes (max for Vercel hobby plan)
 export const maxDuration = 300;
 
 export const { GET, POST, PUT } = serve({
   client: inngest,
-  functions: [runTestSuite, runSecurityScan, runApiTests, runPerformanceTests, runVibetest, runSecurityAgent],
-});
+  functions: [runTestSuite, runSecurityScan, runApiTests, runPerformanceTests, runVibetest, runSecurityAgent, indexDeploymentLogs, cronReindexLogs],
+});

app/api/logs/search/route.ts

@@ -0,0 +1,160 @@
+/**
+ * app/api/logs/search/route.ts
+ *
+ * GET  /api/logs/search?q=<query>[&sandboxId=<id>][&limit=<n>]
+ *   Returns semantically similar log chunks for the authenticated user.
+ *
+ * POST /api/logs/search
+ *   Body: { q: string; sandboxId?: string; limit?: number }
+ *   Same as GET but accepts a JSON body (useful for longer queries).
+ *
+ * POST /api/logs/search?action=trigger
+ *   Body: { sandboxId: string; ttlDays?: number }
+ *   Manually trigger indexing for a sandbox (fires the Inngest event).
+ */
+
+import { NextRequest, NextResponse } from "next/server";
+import { auth } from "@/lib/auth";
+import { searchLogVectors } from "@/lib/db";
+import { inngest } from "@/lib/inngest";
+import { embedQuery } from "@/lib/vector-indexer";
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+
+function sanitiseLimit(raw: string | null | undefined): number {
+  const n = parseInt(raw ?? "10", 10);
+  if (isNaN(n) || n < 1) return 10;
+  if (n > 50) return 50;
+  return n;
+}
+
+// ── Shared search handler ─────────────────────────────────────────────────────
+
+async function handleSearch(opts: {
+  userId: string;
+  query: string;
+  sandboxId?: string;
+  limit: number;
+}) {
+  const { userId, query, sandboxId, limit } = opts;
+
+  if (!query || query.trim().length === 0) {
+    return NextResponse.json(
+      { error: "Query parameter 'q' is required" },
+      { status: 400 }
+    );
+  }
+
+  // Embed the user's natural-language query using Cohere
+  const queryEmbedding = await embedQuery(query.trim());
+
+  // Run the cosine-similarity search
+  const results = await searchLogVectors({
+    userId,
+    queryEmbedding,
+    sandboxId,
+    limit,
+  });
+
+  return NextResponse.json({
+    query,
+    sandboxId: sandboxId ?? null,
+    count: results.length,
+    results: results.map((r) => ({
+      id: r.id,
+      sandboxId: r.sandboxId,
+      logIdRange: { start: r.logIdStart, end: r.logIdEnd },
+      chunkText: r.chunkText,
+      createdAt: r.createdAt,
+    })),
+  });
+}
+
+// ── GET /api/logs/search ──────────────────────────────────────────────────────
+
+export async function GET(req: NextRequest) {
+  const session = await auth();
+  if (!session?.user?.id) {
+    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+  }
+
+  const { searchParams } = new URL(req.url);
+  const query = searchParams.get("q") ?? "";
+  const sandboxId = searchParams.get("sandboxId") ?? undefined;
+  const limit = sanitiseLimit(searchParams.get("limit"));
+
+  try {
+    return await handleSearch({ userId: session.user.id, query, sandboxId, limit });
+  } catch (err) {
+    console.error("[logs/search] GET error:", err);
+    return NextResponse.json(
+      { error: "Search failed", detail: String(err) },
+      { status: 500 }
+    );
+  }
+}
+
+// ── POST /api/logs/search ─────────────────────────────────────────────────────
+
+export async function POST(req: NextRequest) {
+  const session = await auth();
+  if (!session?.user?.id) {
+    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+  }
+
+  // Handle manual index trigger via ?action=trigger
+  const { searchParams } = new URL(req.url);
+  if (searchParams.get("action") === "trigger") {
+    return handleTrigger(req, session.user.id);
+  }
+
+  let body: { q?: string; sandboxId?: string; limit?: number };
+  try {
+    body = await req.json();
+  } catch {
+    return NextResponse.json({ error: "Invalid JSON body" }, { status: 400 });
+  }
+
+  const query = body.q ?? "";
+  const sandboxId = body.sandboxId ?? undefined;
+  const limit = sanitiseLimit(String(body.limit ?? 10));
+
+  try {
+    return await handleSearch({ userId: session.user.id, query, sandboxId, limit });
+  } catch (err) {
+    console.error("[logs/search] POST error:", err);
+    return NextResponse.json(
+      { error: "Search failed", detail: String(err) },
+      { status: 500 }
+    );
+  }
+}
+
+// ── Manual index trigger ──────────────────────────────────────────────────────
+
+async function handleTrigger(req: NextRequest, userId: string) {
+  let body: { sandboxId?: string; ttlDays?: number };
+  try {
+    body = await req.json();
+  } catch {
+    return NextResponse.json({ error: "Invalid JSON body" }, { status: 400 });
+  }
+
+  const { sandboxId, ttlDays = 30 } = body;
+  if (!sandboxId) {
+    return NextResponse.json(
+      { error: "sandboxId is required" },
+      { status: 400 }
+    );
+  }
+
+  await inngest.send({
+    name: "log/index.requested",
+    data: { sandboxId, userId, ttlDays },
+  });
+
+  return NextResponse.json({
+    ok: true,
+    message: `Indexing triggered for sandbox ${sandboxId}`,
+  });
+}

lib/db.ts

@@ -259,6 +259,44 @@ export async function ensureTables(): Promise<void> {
       ON notifications (user_id, created_at DESC)
   `;
 
+  // ── pgvector extension ────────────────────────────────────────────────────
+  // Requires pgvector installed on your Neon project (available by default).
+  await sql`CREATE EXTENSION IF NOT EXISTS vector`;
+
+  // ── Log vector index table ────────────────────────────────────────────────
+  // Stores chunked log text + 1536-dim OpenAI embeddings for semantic search.
+  await sql`
+    CREATE TABLE IF NOT EXISTS deployment_log_vectors (
+      id            BIGSERIAL PRIMARY KEY,
+      sandbox_id    TEXT      NOT NULL,
+      user_id       TEXT      NOT NULL DEFAULT '',
+      log_id_start  BIGINT    NOT NULL,
+      log_id_end    BIGINT    NOT NULL,
+      chunk_text    TEXT      NOT NULL,
+      embedding     vector(1024),
+      created_at    BIGINT    NOT NULL,
+      expires_at    BIGINT    NOT NULL
+    )
+  `;
+
+  await sql`
+    CREATE INDEX IF NOT EXISTS idx_log_vectors_sandbox
+      ON deployment_log_vectors (sandbox_id)
+  `;
+
+  // HNSW index for fast approximate nearest-neighbour search
+  await sql`
+    CREATE INDEX IF NOT EXISTS idx_log_vectors_embedding
+      ON deployment_log_vectors
+      USING hnsw (embedding vector_cosine_ops)
+  `;
+
+  // Index to speed up the retention-pruning query
+  await sql`
+    CREATE INDEX IF NOT EXISTS idx_log_vectors_expires
+      ON deployment_log_vectors (expires_at)
+  `;
+
   tablesReady = true;
 }
 
@@ -309,3 +347,125 @@ export async function createNotification(opts: {
     `;
   } catch { /* non-blocking */ }
 }
+
+// ── pgvector helpers ──────────────────────────────────────────────────────────
+
+export interface LogVector {
+  id: number;
+  sandboxId: string;
+  userId: string;
+  logIdStart: number;
+  logIdEnd: number;
+  chunkText: string;
+  createdAt: number;
+}
+
+/**
+ * Insert a pre-computed embedding chunk into deployment_log_vectors.
+ * Called by the background indexer — not by API routes directly.
+ */
+export async function insertLogVector(opts: {
+  sandboxId: string;
+  userId: string;
+  logIdStart: number;
+  logIdEnd: number;
+  chunkText: string;
+  embedding: number[];
+  /** TTL in milliseconds from now — defaults to 30 days */
+  ttlMs?: number;
+}): Promise<void> {
+  const sql = getDb();
+  const expiresAt = Date.now() + (opts.ttlMs ?? 30 * 24 * 60 * 60 * 1000);
+  await sql`
+    INSERT INTO deployment_log_vectors
+      (sandbox_id, user_id, log_id_start, log_id_end,
+       chunk_text, embedding, created_at, expires_at)
+    VALUES (
+      ${opts.sandboxId},
+      ${opts.userId},
+      ${opts.logIdStart},
+      ${opts.logIdEnd},
+      ${opts.chunkText},
+      ${JSON.stringify(opts.embedding)}::vector,
+      ${Date.now()},
+      ${expiresAt}
+    )
+  `;
+}
+
+/**
+ * Semantic nearest-neighbour search over deployment_log_vectors.
+ * Returns up to `limit` chunks ranked by cosine similarity.
+ */
+export async function searchLogVectors(opts: {
+  userId: string;
+  queryEmbedding: number[];
+  sandboxId?: string;
+  limit?: number;
+}): Promise<LogVector[]> {
+  const sql = getDb();
+  const limit = opts.limit ?? 10;
+  const vec = JSON.stringify(opts.queryEmbedding);
+
+  let rows;
+  if (opts.sandboxId) {
+    rows = await sql`
+      SELECT id, sandbox_id, user_id, log_id_start, log_id_end,
+             chunk_text, created_at
+      FROM   deployment_log_vectors
+      WHERE  user_id    = ${opts.userId}
+        AND  sandbox_id = ${opts.sandboxId}
+        AND  expires_at > ${Date.now()}
+      ORDER BY embedding <=> ${vec}::vector
+      LIMIT  ${limit}
+    `;
+  } else {
+    rows = await sql`
+      SELECT id, sandbox_id, user_id, log_id_start, log_id_end,
+             chunk_text, created_at
+      FROM   deployment_log_vectors
+      WHERE  user_id    = ${opts.userId}
+        AND  expires_at > ${Date.now()}
+      ORDER BY embedding <=> ${vec}::vector
+      LIMIT  ${limit}
+    `;
+  }
+
+  return rows.map((r) => ({
+    id: Number(r.id),
+    sandboxId: r.sandbox_id,
+    userId: r.user_id,
+    logIdStart: Number(r.log_id_start),
+    logIdEnd: Number(r.log_id_end),
+    chunkText: r.chunk_text,
+    createdAt: Number(r.created_at),
+  }));
+}
+
+/**
+ * Delete expired log vectors (retention policy).
+ * Run periodically from the Inngest indexer or a cron route.
+ */
+export async function pruneExpiredLogVectors(): Promise<number> {
+  const sql = getDb();
+  const result = await sql`
+    DELETE FROM deployment_log_vectors
+    WHERE expires_at < ${Date.now()}
+    RETURNING id
+  `;
+  return result.length;
+}
+
+/**
+ * Returns the highest log id already indexed for a given sandbox,
+ * so the indexer knows where to resume.
+ */
+export async function getLastIndexedLogId(sandboxId: string): Promise<number> {
+  const sql = getDb();
+  const rows = await sql`
+    SELECT COALESCE(MAX(log_id_end), 0) AS last_id
+    FROM   deployment_log_vectors
+    WHERE  sandbox_id = ${sandboxId}
+  `;
+  return Number(rows[0]?.last_id ?? 0);
+}

lib/deployer.ts

@@ -1,5 +1,6 @@
 import { Sandbox } from "e2b";
 import { getDb, ensureTables } from "./db";
+import { inngest } from "./inngest";
 
 // Custom template: Node 20 + git + pnpm + serve pre-installed (qg1v6gyvxew6q52r04lp)
 const E2B_TEMPLATE = process.env.E2B_TEMPLATE ?? "secdev-web-runtime";
@@ -177,7 +178,7 @@ export async function startDeployment(
   `;
 
   // Fire-and-forget background deployment pipeline
-  runDeploymentPipeline(sandbox, sandboxId, repoName, publicUrl, repoUrl, branch, options?.envVars)
+  runDeploymentPipeline(sandbox, sandboxId, repoName, publicUrl, repoUrl, branch, userId, options?.envVars)
     .catch(async (err: unknown) => {
       const msg = err instanceof Error ? err.message : String(err);
       const ts = Date.now();
@@ -198,6 +199,7 @@ async function runDeploymentPipeline(
   publicUrl: string,
   repoUrl: string,
   branch: string,
+  userId: string,
   envVars?: Record<string, string>
 ): Promise<void> {
   const sql = getDb();
@@ -395,6 +397,12 @@ async function runDeploymentPipeline(
 
   await setStatus("live");
   log(`🎉 Deployment LIVE → ${publicUrl}`);
+
+  // Trigger background log indexing for semantic search
+  inngest.send({
+    name: "log/index.requested",
+    data: { sandboxId, userId, ttlDays: 30 },
+  }).catch(() => null); // fire-and-forget, never block the deployment
 }
 
 // ── Sandbox management ─────────────────────────────────────────────────────────
@@ -443,5 +451,4 @@ export async function refreshSandboxStatus(
     await sql`UPDATE deployments SET status = 'failed' WHERE sandbox_id = ${sandboxId}`;
     return "failed";
   }
-}
-
+}