engineering/tpmjs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2.2k

fix: separate sync discovery from enrichment and harden Railway executor

Browse source 
Sync system was timing out because discovery endpoints (keyword, changes)
also ran schema extraction (~10-15s per tool). Now discovery is fast
(npm metadata + DB writes only) and a new /api/sync/enrich endpoint
handles schema extraction in time-budgeted chunks.

Railway executor was crashing without restarting due to unhandled promise
rejections, no restart policy, and no health checks. Added crash
protection, graceful shutdown, cache size limits, railway.toml with
ALWAYS restart policy, and upgraded Deno from 1.39 to 2.1.9.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Thomas Davis 2026-02-09 04:49:54 +10:00
parent 9ec8bf2454
commit 198f9f7d1e
10 changed files with 557 additions and 282 deletions
Download patch file Download diff file Expand all files Collapse all files

114
.github/workflows/sync-enrich.yml vendored Normal file
View file

@ -0,0 +1,114 @@
name: Sync Tool Enrichment
on:
schedule:
# Run every 2 minutes
- cron: '*/2 * * * *'
workflow_dispatch:
jobs:
sync-enrich:
runs-on: ubuntu-latest
steps:
- name: Trigger enrichment sync
id: sync
run: |
# Call the sync API and capture response
response=$(curl -X POST "${{ secrets.VERCEL_PRODUCTION_URL }}/api/sync/enrich" \
-H "Authorization: Bearer ${{ secrets.CRON_SECRET }}" \
-H "Content-Type: application/json" \
-f -s -S)
echo "Response: $response"
# Extract data using jq
enriched=$(echo "$response" | jq -r '.data.enriched')
discovered=$(echo "$response" | jq -r '.data.discovered')
skipped=$(echo "$response" | jq -r '.data.skipped')
errors=$(echo "$response" | jq -r '.data.errors')
durationMs=$(echo "$response" | jq -r '.data.durationMs')
# Extract and display error messages
errorMessages=$(echo "$response" | jq -r '.data.errorMessages[]?' 2>/dev/null || echo "")
if [ -n "$errorMessages" ]; then
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
echo "⚠️ ENRICHMENT ERRORS ($errors total):"
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
echo "$response" | jq -r '.data.errorMessages[]?' | while IFS= read -r error; do
echo " • $error"
done
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
fi
# Set outputs for Discord notification
echo "enriched=$enriched" >> $GITHUB_OUTPUT
echo "discovered=$discovered" >> $GITHUB_OUTPUT
echo "skipped=$skipped" >> $GITHUB_OUTPUT
echo "errors=$errors" >> $GITHUB_OUTPUT
echo "durationMs=$durationMs" >> $GITHUB_OUTPUT
# Store error messages for Discord (first 3, truncated)
if [ "$errors" -gt 0 ]; then
errorSummary=$(echo "$response" | jq -r '.data.errorMessages[0:3]? | join("\n• ")' 2>/dev/null || echo "")
if [ -n "$errorSummary" ]; then
echo "• $errorSummary" > /tmp/error_summary.txt
fi
fi
# Determine status emoji
if [ "$errors" -gt 0 ]; then
echo "status_emoji=⚠️" >> $GITHUB_OUTPUT
echo "status_color=16776960" >> $GITHUB_OUTPUT # Yellow
else
echo "status_emoji=✅" >> $GITHUB_OUTPUT
echo "status_color=5763719" >> $GITHUB_OUTPUT # Green
fi
- name: Send Discord notification
if: always()
run: |
# Format duration
duration_sec=$(echo "scale=2; ${{ steps.sync.outputs.durationMs }} / 1000" | bc)
# Build Discord payload using jq for proper JSON escaping
error_text=""
if [ -f /tmp/error_summary.txt ] && [ ${{ steps.sync.outputs.errors }} -gt 0 ]; then
error_text=$(cat /tmp/error_summary.txt | head -c 800)
fi
# Build fields array dynamically
base_fields='[
{ "name": "🔧 Enriched", "value": "${{ steps.sync.outputs.enriched }}", "inline": true },
{ "name": "🔍 Discovered", "value": "${{ steps.sync.outputs.discovered }}", "inline": true },
{ "name": "⏭️ Skipped", "value": "${{ steps.sync.outputs.skipped }}", "inline": true },
{ "name": "❌ Errors", "value": "${{ steps.sync.outputs.errors }}", "inline": true },
{ "name": "⏱️ Duration", "value": "'"${duration_sec}s"'", "inline": true },
{ "name": "🔗 Run", "value": "[View Logs](https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }})", "inline": true }
]'
# Create payload with dynamic fields
payload=$(jq -n \
--arg title "${{ steps.sync.outputs.status_emoji }} Tool Enrichment Sync" \
--argjson color ${{ steps.sync.outputs.status_color }} \
--argjson baseFields "$base_fields" \
--arg error_text "$error_text" \
--arg timestamp "$(date -u +%Y-%m-%dT%H:%M:%S.000Z)" \
'
{
embeds: [{
title: $title,
color: $color,
fields: (
$baseFields +
(if $error_text != "" then [{ name: "🔍 Error Details", value: ("```\n" + $error_text + "\n```"), inline: false }] else [] end)
),
timestamp: $timestamp
}]
}')
# Send to Discord
curl -X POST "${{ secrets.DISCORD_WEBHOOK }}" \
-H "Content-Type: application/json" \
-d "$payload"

10
apps/railway-executor/Dockerfile
View file

@ -1,9 +1,9 @@
# Use official Deno image
FROM denoland/deno:1.39.0
# Use latest Deno LTS image for stability
FROM denoland/deno:2.1.9
# Install OpenSSH client for tools that need SSH access (e.g., exe-dev)
USER root
RUN apt-get update && apt-get install -y openssh-client && rm -rf /var/lib/apt/lists/*
RUN apt-get update && apt-get install -y --no-install-recommends openssh-client curl && rm -rf /var/lib/apt/lists/*
# Set working directory
WORKDIR /app
@ -19,5 +19,9 @@ RUN chmod +x start.sh
# Expose port (Railway will set PORT env var)
EXPOSE 3002
# Docker-level health check as a fallback
HEALTHCHECK --interval=30s --timeout=10s --start-period=15s --retries=3 \
CMD curl -f http://localhost:${PORT:-3002}/health || exit 1
# Run startup script that fixes permissions then starts Deno as deno user
CMD ["./start.sh"]

10
apps/railway-executor/railway.toml Normal file
View file

@ -0,0 +1,10 @@
[build]
builder = "DOCKERFILE"
dockerfilePath = "Dockerfile"
[deploy]
startCommand = "./start.sh"
healthcheckPath = "/health"
healthcheckTimeout = 30
restartPolicyType = "ALWAYS"
restartPolicyMaxRetries = -1

57
apps/railway-executor/server.ts
View file

@ -6,9 +6,25 @@
// Import zod-to-json-schema for Zod v3 support
import { zodToJsonSchema } from 'https://esm.sh/zod-to-json-schema@3.25.0';
// ─── Crash Protection ───────────────────────────────────────────────────────
// Catch unhandled promise rejections so they don't crash the process
globalThis.addEventListener('unhandledrejection', (event) => {
event.preventDefault();
console.error('⚠️ Unhandled promise rejection (caught, process continues):', event.reason);
});
// Catch uncaught errors
globalThis.addEventListener('error', (event) => {
console.error('⚠️ Uncaught error (caught, process continues):', event.error || event.message);
event.preventDefault();
});
// Cache TTL: 2 minutes
const CACHE_TTL_MS = 2 * 60 * 1000;
// Max cache entries to prevent unbounded memory growth
const MAX_CACHE_SIZE = 200;
// Cache entry with expiration
interface CacheEntry {
// biome-ignore lint/suspicious/noExplicitAny: Tool types are dynamic and vary by package
@ -41,6 +57,16 @@ function getCachedModule(cacheKey: string): CacheEntry | null {
*/
// biome-ignore lint/suspicious/noExplicitAny: Tool types are dynamic and vary by package
function setCachedModule(cacheKey: string, module: any, isFactory: boolean): void {
// Evict oldest entries if cache is full
if (moduleCache.size >= MAX_CACHE_SIZE) {
const entriesToEvict = Math.max(1, Math.floor(MAX_CACHE_SIZE * 0.2)); // Evict 20%
const keys = Array.from(moduleCache.keys());
for (let i = 0; i < entriesToEvict && i < keys.length; i++) {
moduleCache.delete(keys[i]);
}
console.log(`🗑️ Evicted ${entriesToEvict} cache entries (cache was full at ${MAX_CACHE_SIZE})`);
}
moduleCache.set(cacheKey, {
module,
expiresAt: Date.now() + CACHE_TTL_MS,
@ -845,6 +871,9 @@ async function listExports(req: Request): Promise<Response> {
}
}
// Track startup time for uptime reporting
const startedAt = Date.now();
/**
* Health check
*/
@ -852,7 +881,9 @@ function health(): Response {
return Response.json({
status: 'ok',
timestamp: new Date().toISOString(),
uptimeSeconds: Math.floor((Date.now() - startedAt) / 1000),
cacheSize: moduleCache.size,
maxCacheSize: MAX_CACHE_SIZE,
denoVersion: Deno.version.deno,
v8Version: Deno.version.v8,
httpImports: true,
@ -893,9 +924,15 @@ function clearCache(): Response {
}
/**
* Main request handler
* Main request handler — wrapped with crash protection so no single request
* can take down the process.
*/
async function handler(req: Request): Promise<Response> {
// Reject requests during shutdown
if (isShuttingDown) {
return new Response('Service shutting down', { status: 503 });
}
const url = new URL(req.url);
// Add CORS headers
@ -946,6 +983,24 @@ async function handler(req: Request): Promise<Response> {
}
}
// ─── Graceful Shutdown ──────────────────────────────────────────────────────
let isShuttingDown = false;
function handleShutdown(signal: string) {
if (isShuttingDown) return;
isShuttingDown = true;
console.log(`\n🛑 Received ${signal}, shutting down gracefully...`);
moduleCache.clear();
// Give in-flight requests a moment to complete
setTimeout(() => {
console.log('👋 Goodbye');
Deno.exit(0);
}, 5000);
}
Deno.addSignalListener('SIGTERM', () => handleShutdown('SIGTERM'));
Deno.addSignalListener('SIGINT', () => handleShutdown('SIGINT'));
// Start server
const port = Number.parseInt(Deno.env.get('PORT') || '3002', 10);

1
apps/railway-executor/start.sh
View file

@ -4,4 +4,5 @@ mkdir -p /tmp/deno-cache
chown -R deno:deno /tmp/deno-cache
# Switch to deno user and run the server
# Deno 2.x: --allow-net, --allow-env, --allow-read, --allow-write, --allow-run for tool execution
exec su deno -c "deno run --allow-net --allow-env --allow-read --allow-write --allow-run server.ts"

137
apps/web/src/app/api/sync/changes/route.ts
View file

@ -1,30 +1,23 @@
import { prisma } from '@tpmjs/db';
import { fetchChanges, fetchLatestPackageWithMetadata } from '@tpmjs/npm-client';
import type { TpmjsToolDefinition } from '@tpmjs/types/tpmjs';
import { validateTpmjsField } from '@tpmjs/types/tpmjs';
import { type NextRequest, NextResponse } from 'next/server';
import { env } from '~/env';
import { performHealthCheck } from '~/lib/health-check/health-check-service';
import {
convertJsonSchemaToParameters,
extractToolSchema,
listToolExports,
} from '~/lib/schema-extraction';
export const runtime = 'nodejs';
export const dynamic = 'force-dynamic';
export const maxDuration = 300; // 5 minutes max for cron jobs
export const maxDuration = 60;
/**
* POST /api/sync/changes
* Sync tools from NPM changes feed
* Discovery-only sync: monitors NPM changes feed, upserts packages and tools.
* Does NOT call the executor for schema extraction or health checks — that's handled by /api/sync/enrich.
*
* This endpoint is called by Vercel Cron (every 2 minutes)
* Called by Vercel Cron (every 4 hours) or GitHub Actions.
* Requires Authorization: Bearer <CRON_SECRET>
*/
// biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Complex but straightforward CRUD operation
export async function POST(request: NextRequest) {
// Verify cron secret for security
const authHeader = request.headers.get('authorization');
const token = authHeader?.replace('Bearer ', '');
@ -39,7 +32,6 @@ export async function POST(request: NextRequest) {
const errorMessages: string[] = [];
try {
// Get last checkpoint
const checkpoint = await prisma.syncCheckpoint.findUnique({
where: { source: 'changes-feed' },
});
@ -48,46 +40,37 @@ export async function POST(request: NextRequest) {
? String((checkpoint.checkpoint as { lastSeq?: string })?.lastSeq || '0')
: '0';
// Fetch changes from NPM (limit to 30 per run to allow time for schema extraction)
// Increased limit since we no longer spend time on schema extraction
const changesResult = await fetchChanges({
since: lastSeq,
limit: 30,
limit: 100,
includeDocs: false,
});
// Process each change
for (const change of changesResult.results) {
try {
// Fetch full package metadata with README
const pkg = await fetchLatestPackageWithMetadata(change.id);
// Skip if package not found
if (!pkg) {
skipped++;
continue;
}
// Check if package has tpmjs field
if (!pkg.tpmjs) {
skipped++;
continue;
}
// Validate tpmjs field (supports both new multi-tool and legacy formats)
const validation = validateTpmjsField(pkg.tpmjs);
if (!validation.valid || !validation.packageData || !validation.tools) {
if (!validation.valid || !validation.packageData) {
skipped++;
continue;
}
// Log auto-migration from legacy format
if (validation.wasLegacyFormat) {
console.log(`Auto-migrated legacy package: ${pkg.name}`);
}
// Extract repository URL and GitHub stars
const githubStars: number | null = null;
// Upsert Package record
const packageRecord = await prisma.package.upsert({
where: { npmPackageName: pkg.name },
@ -109,8 +92,8 @@ export async function POST(request: NextRequest) {
tier: validation.tier || 'minimal',
discoveryMethod: 'changes-feed',
isOfficial: pkg.keywords?.includes('tpmjs') || false,
npmDownloadsLastMonth: 0, // Will be updated by metrics sync
githubStars: githubStars,
npmDownloadsLastMonth: 0,
githubStars: null,
},
update: {
npmVersion: pkg.version,
@ -131,53 +114,30 @@ export async function POST(request: NextRequest) {
},
});
// Get existing tools for this package
// For auto-discovery packages, skip tool creation — enrichment will handle it
if (validation.needsAutoDiscovery) {
console.log(
`Package ${pkg.name} needs auto-discovery — enrichment will handle tool creation`
);
processed++;
continue;
}
// Upsert tools from the tpmjs.tools array (manual discovery only)
const toolsToProcess = validation.tools || [];
const existingTools = await prisma.tool.findMany({
where: { packageId: packageRecord.id },
});
// Determine the tools to process
let toolsToProcess: TpmjsToolDefinition[] = validation.tools || [];
let toolDiscoverySource: 'auto' | 'manual' = 'manual';
// If tools need auto-discovery, call the executor to list exports
if (validation.needsAutoDiscovery) {
console.log(`Auto-discovering tools for ${pkg.name}...`);
const exportsResult = await listToolExports(pkg.name, pkg.version, null);
if (exportsResult.success) {
// Convert discovered tools to TpmjsToolDefinition format
toolsToProcess = exportsResult.tools
.filter((t) => t.isValidTool)
.map((t) => ({
name: t.name,
description: t.description,
parameters: undefined,
returns: undefined,
aiAgent: undefined,
}));
toolDiscoverySource = 'auto';
console.log(
`Auto-discovered ${toolsToProcess.length} tools for ${pkg.name}: ${toolsToProcess.map((t) => t.name).join(', ')}`
);
} else {
console.log(`Failed to auto-discover tools for ${pkg.name}: ${exportsResult.error}`);
// Skip this package if we can't discover tools
skipped++;
continue;
}
}
// Upsert each tool
for (const toolDef of toolsToProcess) {
// Get tool name from validated schema
const toolName = toolDef.name;
if (!toolName) {
console.warn(`Skipping tool without name in ${pkg.name}`);
continue;
}
const upsertedTool = await prisma.tool.upsert({
await prisma.tool.upsert({
where: {
packageId_name: {
packageId: packageRecord.id,
@ -194,10 +154,9 @@ export async function POST(request: NextRequest) {
returns: toolDef.returns ? (toolDef.returns as any) : undefined,
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
aiAgent: toolDef.aiAgent ? (toolDef.aiAgent as any) : undefined,
qualityScore: null, // Will be calculated by metrics sync
// Schema will be extracted below
qualityScore: null,
schemaSource: toolDef.parameters ? 'author' : null,
toolDiscoverySource,
toolDiscoverySource: 'manual',
},
update: {
description: toolDef.description || undefined,
@ -207,52 +166,9 @@ export async function POST(request: NextRequest) {
returns: toolDef.returns ? (toolDef.returns as any) : undefined,
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
aiAgent: toolDef.aiAgent ? (toolDef.aiAgent as any) : undefined,
toolDiscoverySource,
toolDiscoverySource: 'manual',
},
});
// Extract schema synchronously from executor
const schemaResult = await extractToolSchema(pkg.name, toolName, pkg.version, null);
if (schemaResult.success) {
// Update tool with extracted schema (and description if not provided)
await prisma.tool.update({
where: { id: upsertedTool.id },
data: {
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
inputSchema: schemaResult.inputSchema as any,
// Also update parameters array for backward compatibility
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
parameters: convertJsonSchemaToParameters(schemaResult.inputSchema) as any,
schemaSource: 'extracted',
schemaExtractedAt: new Date(),
// Update description if not provided by author
...(!toolDef.description && schemaResult.description
? { description: schemaResult.description }
: {}),
},
});
console.log(`Schema extracted for ${pkg.name}/${toolName}`);
} else {
// Extraction failed - mark schema source appropriately
console.log(
`Schema extraction failed for ${pkg.name}/${toolName}: ${schemaResult.error}`
);
await prisma.tool.update({
where: { id: upsertedTool.id },
data: {
schemaSource: toolDef.parameters ? 'author' : null,
},
});
}
// Trigger health check (non-blocking) for execution testing
performHealthCheck(upsertedTool.id, 'sync').catch((err) => {
console.error(
`Health check failed for ${pkg.name}/${toolName} (${upsertedTool.id}):`,
err
);
});
}
// Delete orphaned tools (tools removed from package.json)
@ -278,7 +194,6 @@ export async function POST(request: NextRequest) {
}
}
// Update checkpoint with new sequence
await prisma.syncCheckpoint.upsert({
where: { source: 'changes-feed' },
create: {
@ -296,7 +211,6 @@ export async function POST(request: NextRequest) {
},
});
// Log sync operation
await prisma.syncLog.create({
data: {
source: 'changes-feed',
@ -330,7 +244,6 @@ export async function POST(request: NextRequest) {
} catch (error) {
console.error('Changes feed sync failed:', error);
// Log failed sync
await prisma.syncLog.create({
data: {
source: 'changes-feed',

261
apps/web/src/app/api/sync/enrich/route.ts Normal file
View file

@ -0,0 +1,261 @@
import { prisma } from '@tpmjs/db';
import { type NextRequest, NextResponse } from 'next/server';
import { env } from '~/env';
import { performHealthCheck } from '~/lib/health-check/health-check-service';
import {
convertJsonSchemaToParameters,
extractToolSchema,
listToolExports,
} from '~/lib/schema-extraction';
export const runtime = 'nodejs';
export const dynamic = 'force-dynamic';
export const maxDuration = 60;
const TIME_BUDGET_MS = 45_000; // Stop starting new work after 45s (leaves 15s buffer)
const RETRY_COOLDOWN_MS = 60 * 60 * 1000; // 1 hour before retrying failed extractions
/**
* POST /api/sync/enrich
* Enrichment queue processor: extracts schemas and runs health checks for tools
* that haven't been enriched yet. Also handles auto-discovery for packages with no tools.
*
* Called by Vercel Cron (every 2 minutes) or GitHub Actions.
* Requires Authorization: Bearer <CRON_SECRET>
*/
// biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Complex but straightforward queue processing
export async function POST(request: NextRequest) {
const authHeader = request.headers.get('authorization');
const token = authHeader?.replace('Bearer ', '');
if (env.CRON_SECRET && token !== env.CRON_SECRET) {
return NextResponse.json({ success: false, error: 'Unauthorized' }, { status: 401 });
}
const startTime = Date.now();
let enriched = 0;
let discovered = 0;
let errors = 0;
let skipped = 0;
const errorMessages: string[] = [];
try {
const now = new Date();
const retryCutoff = new Date(now.getTime() - RETRY_COOLDOWN_MS);
// Phase 1: Auto-discover tools for packages that have 0 tools
const packagesNeedingDiscovery = await prisma.package.findMany({
where: {
tools: { none: {} },
},
take: 5,
});
for (const pkg of packagesNeedingDiscovery) {
if (Date.now() - startTime > TIME_BUDGET_MS) {
console.log('Time budget exceeded during auto-discovery phase, stopping');
break;
}
try {
console.log(`Auto-discovering tools for ${pkg.npmPackageName}...`);
const exportsResult = await listToolExports(
pkg.npmPackageName,
pkg.npmVersion,
pkg.env as Record<string, unknown> | null
);
if (!exportsResult.success) {
console.log(
`Failed to auto-discover tools for ${pkg.npmPackageName}: ${exportsResult.error}`
);
errors++;
errorMessages.push(
`Auto-discovery failed for ${pkg.npmPackageName}: ${exportsResult.error}`
);
continue;
}
const validTools = exportsResult.tools.filter((t) => t.isValidTool);
for (const tool of validTools) {
await prisma.tool.upsert({
where: {
packageId_name: {
packageId: pkg.id,
name: tool.name,
},
},
create: {
packageId: pkg.id,
name: tool.name,
description: tool.description || 'No description provided',
qualityScore: null,
schemaSource: null,
toolDiscoverySource: 'auto',
},
update: {
description: tool.description || undefined,
toolDiscoverySource: 'auto',
},
});
}
console.log(
`Auto-discovered ${validTools.length} tools for ${pkg.npmPackageName}: ${validTools.map((t) => t.name).join(', ')}`
);
discovered++;
} catch (error) {
errors++;
const errorMsg = `Auto-discovery error for ${pkg.npmPackageName}: ${error instanceof Error ? error.message : 'Unknown error'}`;
errorMessages.push(errorMsg);
console.error(errorMsg);
}
}
// Phase 2: Enrich tools that need schema extraction
const toolsToEnrich = await prisma.tool.findMany({
where: {
schemaSource: null,
OR: [
{ schemaExtractionAttemptAt: null },
{ schemaExtractionAttemptAt: { lt: retryCutoff } },
],
},
include: { package: true },
take: 10, // Fetch a few more than we'll likely process
orderBy: { createdAt: 'asc' }, // Oldest first
});
for (const tool of toolsToEnrich) {
if (Date.now() - startTime > TIME_BUDGET_MS) {
console.log('Time budget exceeded during enrichment phase, stopping');
skipped += toolsToEnrich.length - enriched;
break;
}
try {
// Mark attempt time before starting (prevents concurrent processing)
await prisma.tool.update({
where: { id: tool.id },
data: { schemaExtractionAttemptAt: now },
});
console.log(`Extracting schema for ${tool.package.npmPackageName}/${tool.name}...`);
const schemaResult = await extractToolSchema(
tool.package.npmPackageName,
tool.name,
tool.package.npmVersion,
tool.package.env as Record<string, unknown> | null
);
if (schemaResult.success) {
await prisma.tool.update({
where: { id: tool.id },
data: {
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
inputSchema: schemaResult.inputSchema as any,
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
parameters: convertJsonSchemaToParameters(schemaResult.inputSchema) as any,
schemaSource: 'extracted',
schemaExtractedAt: new Date(),
schemaExtractionError: null,
// Update description if not already set meaningfully
...(tool.description === 'No description provided' && schemaResult.description
? { description: schemaResult.description }
: {}),
},
});
console.log(`Schema extracted for ${tool.package.npmPackageName}/${tool.name}`);
// Trigger health check after successful extraction
performHealthCheck(tool.id, 'enrich').catch((err) => {
console.error(
`Health check failed for ${tool.package.npmPackageName}/${tool.name}:`,
err
);
});
enriched++;
} else {
await prisma.tool.update({
where: { id: tool.id },
data: {
schemaExtractionError: schemaResult.error,
},
});
console.log(
`Schema extraction failed for ${tool.package.npmPackageName}/${tool.name}: ${schemaResult.error}`
);
errors++;
errorMessages.push(`${tool.package.npmPackageName}/${tool.name}: ${schemaResult.error}`);
}
} catch (error) {
errors++;
const errorMsg = `${tool.package.npmPackageName}/${tool.name}: ${error instanceof Error ? error.message : 'Unknown error'}`;
errorMessages.push(errorMsg);
console.error(`Enrichment error: ${errorMsg}`);
}
}
// Log sync operation
await prisma.syncLog.create({
data: {
source: 'enrichment',
status: errors > 0 ? 'partial' : 'success',
processed: enriched + discovered,
skipped,
errors,
message:
errors > 0
? `Enriched ${enriched} tools, discovered ${discovered} packages. Errors: ${errorMessages.slice(0, 3).join('; ')}`
: `Enriched ${enriched} tools, discovered ${discovered} packages`,
metadata: {
durationMs: Date.now() - startTime,
enriched,
discovered,
toolsInQueue: toolsToEnrich.length,
packagesNeedingDiscovery: packagesNeedingDiscovery.length,
},
},
});
return NextResponse.json({
success: true,
data: {
enriched,
discovered,
skipped,
errors,
durationMs: Date.now() - startTime,
errorMessages: errorMessages.slice(0, 5),
},
});
} catch (error) {
console.error('Enrichment sync failed:', error);
await prisma.syncLog.create({
data: {
source: 'enrichment',
status: 'error',
processed: enriched + discovered,
skipped,
errors: errors + 1,
message: error instanceof Error ? error.message : 'Unknown error',
metadata: {
durationMs: Date.now() - startTime,
},
},
});
return NextResponse.json(
{
success: false,
error: 'Enrichment failed',
message: error instanceof Error ? error.message : 'Unknown error',
},
{ status: 500 }
);
}
}

149
apps/web/src/app/api/sync/keyword/route.ts
View file

@ -1,30 +1,23 @@
import { prisma } from '@tpmjs/db';
import { fetchLatestPackageWithMetadata, searchByKeyword } from '@tpmjs/npm-client';
import type { TpmjsToolDefinition } from '@tpmjs/types/tpmjs';
import { validateTpmjsField } from '@tpmjs/types/tpmjs';
import { type NextRequest, NextResponse } from 'next/server';
import { env } from '~/env';
import { performHealthCheck } from '~/lib/health-check/health-check-service';
import {
convertJsonSchemaToParameters,
extractToolSchema,
listToolExports,
} from '~/lib/schema-extraction';
export const runtime = 'nodejs';
export const dynamic = 'force-dynamic';
export const maxDuration = 300; // 5 minutes max for cron jobs
export const maxDuration = 60;
/**
* POST /api/sync/keyword
* Sync tools by searching NPM for 'tpmjs' keyword
* Discovery-only sync: searches NPM for 'tpmjs' keyword, upserts packages and tools.
* Does NOT call the executor for schema extraction or health checks — that's handled by /api/sync/enrich.
*
* This endpoint is called by Vercel Cron (every 15 minutes)
* Called by Vercel Cron (every 6 hours) or GitHub Actions.
* Requires Authorization: Bearer <CRON_SECRET>
*/
// biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Complex but straightforward CRUD operation
export async function POST(request: NextRequest) {
// Verify cron secret for security
const authHeader = request.headers.get('authorization');
const token = authHeader?.replace('Bearer ', '');
@ -40,19 +33,15 @@ export async function POST(request: NextRequest) {
const skippedPackages: Array<{ name: string; author: string; reason: string }> = [];
try {
// Search for packages with 'tpmjs' keyword
const searchResults = await searchByKeyword({
keyword: 'tpmjs',
size: 250, // Get up to 250 packages per sync
size: 250,
});
// Process each package
for (const result of searchResults) {
try {
// Fetch full package metadata with README
const pkg = await fetchLatestPackageWithMetadata(result.package.name);
// Skip if package not found
if (!pkg) {
skipped++;
skippedPackages.push({
@ -63,7 +52,6 @@ export async function POST(request: NextRequest) {
continue;
}
// Extract author name
const authorName =
typeof pkg.author === 'string'
? pkg.author
@ -71,11 +59,10 @@ export async function POST(request: NextRequest) {
? pkg.author.name
: 'unknown';
// Check if package has tpmjs field - if not, we'll auto-discover with defaults
// Validate tpmjs field or use auto-discovery defaults
let validation: ReturnType<typeof validateTpmjsField>;
if (!pkg.tpmjs) {
// No tpmjs field - use auto-discovery with default category
console.log(
`Package ${pkg.name} has tpmjs keyword but no tpmjs field - using auto-discovery`
);
@ -83,14 +70,13 @@ export async function POST(request: NextRequest) {
valid: true,
tier: 'minimal',
packageData: {
category: 'utilities', // Default category for keyword-only packages
category: 'utilities',
},
tools: [],
needsAutoDiscovery: true,
wasLegacyFormat: false,
};
} else {
// Validate tpmjs field (supports both new multi-tool and legacy formats)
validation = validateTpmjsField(pkg.tpmjs);
if (!validation.valid || !validation.packageData) {
skipped++;
@ -102,18 +88,14 @@ export async function POST(request: NextRequest) {
continue;
}
// Log auto-migration from legacy format
if (validation.wasLegacyFormat) {
console.log(`Auto-migrated legacy package: ${pkg.name}`);
}
}
// Extract packageData (guaranteed to exist at this point)
// biome-ignore lint/style/noNonNullAssertion: guaranteed by validation check above
const packageData = validation.packageData!;
// Extract repository URL and GitHub stars
const githubStars: number | null = null;
// Upsert Package record
const packageRecord = await prisma.package.upsert({
where: { npmPackageName: pkg.name },
@ -135,8 +117,8 @@ export async function POST(request: NextRequest) {
tier: validation.tier || 'minimal',
discoveryMethod: 'keyword',
isOfficial: pkg.keywords?.includes('tpmjs') || false,
npmDownloadsLastMonth: 0, // Will be updated by metrics sync
githubStars: githubStars,
npmDownloadsLastMonth: 0,
githubStars: null,
},
update: {
npmVersion: pkg.version,
@ -157,58 +139,30 @@ export async function POST(request: NextRequest) {
},
});
// Get existing tools for this package
// For auto-discovery packages, skip tool creation — enrichment will handle it
if (validation.needsAutoDiscovery) {
console.log(
`Package ${pkg.name} needs auto-discovery — enrichment will handle tool creation`
);
processed++;
continue;
}
// Upsert tools from the tpmjs.tools array (manual discovery only)
const toolsToProcess = validation.tools || [];
const existingTools = await prisma.tool.findMany({
where: { packageId: packageRecord.id },
});
// Determine the tools to process
let toolsToProcess: TpmjsToolDefinition[] = validation.tools || [];
let toolDiscoverySource: 'auto' | 'manual' = 'manual';
// If tools need auto-discovery, call the executor to list exports
if (validation.needsAutoDiscovery) {
console.log(`Auto-discovering tools for ${pkg.name}...`);
const exportsResult = await listToolExports(pkg.name, pkg.version, null);
if (exportsResult.success) {
// Convert discovered tools to TpmjsToolDefinition format
toolsToProcess = exportsResult.tools
.filter((t) => t.isValidTool)
.map((t) => ({
name: t.name,
description: t.description,
parameters: undefined,
returns: undefined,
aiAgent: undefined,
}));
toolDiscoverySource = 'auto';
console.log(
`Auto-discovered ${toolsToProcess.length} tools for ${pkg.name}: ${toolsToProcess.map((t) => t.name).join(', ')}`
);
} else {
console.log(`Failed to auto-discover tools for ${pkg.name}: ${exportsResult.error}`);
// Skip this package if we can't discover tools
skipped++;
skippedPackages.push({
name: pkg.name,
author: authorName,
reason: `auto-discovery failed: ${exportsResult.error}`,
});
continue;
}
}
// Upsert each tool
for (const toolDef of toolsToProcess) {
// Get tool name from validated schema
const toolName = toolDef.name;
if (!toolName) {
console.warn(`Skipping tool without name in ${pkg.name}`);
continue;
}
const upsertedTool = await prisma.tool.upsert({
await prisma.tool.upsert({
where: {
packageId_name: {
packageId: packageRecord.id,
@ -225,10 +179,9 @@ export async function POST(request: NextRequest) {
returns: toolDef.returns ? (toolDef.returns as any) : undefined,
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
aiAgent: toolDef.aiAgent ? (toolDef.aiAgent as any) : undefined,
qualityScore: null, // Will be calculated by metrics sync
// Schema will be extracted below
qualityScore: null,
schemaSource: toolDef.parameters ? 'author' : null,
toolDiscoverySource,
toolDiscoverySource: 'manual',
},
update: {
description: toolDef.description || undefined,
@ -238,52 +191,9 @@ export async function POST(request: NextRequest) {
returns: toolDef.returns ? (toolDef.returns as any) : undefined,
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
aiAgent: toolDef.aiAgent ? (toolDef.aiAgent as any) : undefined,
toolDiscoverySource,
toolDiscoverySource: 'manual',
},
});
// Extract schema synchronously from executor
const schemaResult = await extractToolSchema(pkg.name, toolName, pkg.version, null);
if (schemaResult.success) {
// Update tool with extracted schema (and description if not provided)
await prisma.tool.update({
where: { id: upsertedTool.id },
data: {
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
inputSchema: schemaResult.inputSchema as any,
// Also update parameters array for backward compatibility
// biome-ignore lint/suspicious/noExplicitAny: Prisma Json type compatibility workaround
parameters: convertJsonSchemaToParameters(schemaResult.inputSchema) as any,
schemaSource: 'extracted',
schemaExtractedAt: new Date(),
// Update description if not provided by author
...(!toolDef.description && schemaResult.description
? { description: schemaResult.description }
: {}),
},
});
console.log(`Schema extracted for ${pkg.name}/${toolName}`);
} else {
// Extraction failed - mark schema source appropriately
console.log(
`Schema extraction failed for ${pkg.name}/${toolName}: ${schemaResult.error}`
);
await prisma.tool.update({
where: { id: upsertedTool.id },
data: {
schemaSource: toolDef.parameters ? 'author' : null,
},
});
}
// Trigger health check (non-blocking) for execution testing
performHealthCheck(upsertedTool.id, 'sync').catch((err) => {
console.error(
`Health check failed for ${pkg.name}/${toolName} (${upsertedTool.id}):`,
err
);
});
}
// Delete orphaned tools (tools removed from package.json)
@ -309,7 +219,6 @@ export async function POST(request: NextRequest) {
}
}
// Update checkpoint with last run timestamp
await prisma.syncCheckpoint.upsert({
where: { source: 'keyword-search' },
create: {
@ -327,7 +236,6 @@ export async function POST(request: NextRequest) {
},
});
// Log sync operation
await prisma.syncLog.create({
data: {
source: 'keyword-search',
@ -354,14 +262,13 @@ export async function POST(request: NextRequest) {
errors,
packagesFound: searchResults.length,
durationMs: Date.now() - startTime,
errorMessages: errorMessages.slice(0, 5), // Include first 5 error messages
skippedPackages: skippedPackages, // Include all skipped package names
errorMessages: errorMessages.slice(0, 5),
skippedPackages: skippedPackages,
},
});
} catch (error) {
console.error('Keyword search sync failed:', error);
// Log failed sync
await prisma.syncLog.create({
data: {
source: 'keyword-search',

96
apps/web/src/app/api/sync/metrics/route.ts
View file

@ -7,16 +7,17 @@ export const runtime = 'nodejs';
export const dynamic = 'force-dynamic';
export const maxDuration = 300; // 5 minutes max for cron jobs
const BATCH_SIZE = 5;
/**
* POST /api/sync/metrics
* Update download stats and quality scores for all packages and tools
*
* This endpoint is called by Vercel Cron (every hour)
* This endpoint is called by Vercel Cron (daily)
* Requires Authorization: Bearer <CRON_SECRET>
*/
// biome-ignore lint/complexity/noExcessiveCognitiveComplexity: Complex but straightforward CRUD operation
export async function POST(request: NextRequest) {
// Verify cron secret for security
const authHeader = request.headers.get('authorization');
const token = authHeader?.replace('Bearer ', '');
@ -31,62 +32,69 @@ export async function POST(request: NextRequest) {
const errorMessages: string[] = [];
try {
// Get all packages with their tools from database
const packages = await prisma.package.findMany({
include: {
tools: true,
},
});
// Process each package
for (const pkg of packages) {
try {
// Fetch download stats from NPM (package-level metric)
const downloads = await fetchDownloadStats(pkg.npmPackageName);
// Process packages in batches of BATCH_SIZE concurrently
for (let i = 0; i < packages.length; i += BATCH_SIZE) {
const batch = packages.slice(i, i + BATCH_SIZE);
// Fetch GitHub stars if repository is available
const githubStars = await fetchGitHubStarsFromRepository(
pkg.npmRepository as { type?: string; url?: string } | string | null
);
const results = await Promise.allSettled(
batch.map(async (pkg) => {
// Fetch downloads and GitHub stars in parallel for each package
const [downloads, githubStars] = await Promise.all([
fetchDownloadStats(pkg.npmPackageName),
fetchGitHubStarsFromRepository(
pkg.npmRepository as { type?: string; url?: string } | string | null
),
]);
// Update package metrics
await prisma.package.update({
where: { id: pkg.id },
data: {
npmDownloadsLastMonth: downloads,
githubStars,
},
});
// Calculate and update quality score for each tool in this package
for (const tool of pkg.tools) {
const qualityScore = calculateQualityScore({
tier: pkg.tier, // Tier is at package level
downloads, // Package downloads
githubStars, // Use freshly fetched stars
hasParameters: !!tool.parameters,
hasReturns: !!tool.returns,
hasAiAgent: !!tool.aiAgent,
});
await prisma.tool.update({
where: { id: tool.id },
await prisma.package.update({
where: { id: pkg.id },
data: {
qualityScore,
npmDownloadsLastMonth: downloads,
githubStars,
},
});
}
processed++;
} catch (error) {
errors++;
const errorMsg = `Failed to process ${pkg.npmPackageName}: ${error instanceof Error ? error.message : 'Unknown error'}`;
errorMessages.push(errorMsg);
console.error(errorMsg);
// Calculate and update quality score for each tool in this package
for (const tool of pkg.tools) {
const qualityScore = calculateQualityScore({
tier: pkg.tier,
downloads,
githubStars,
hasParameters: !!tool.parameters,
hasReturns: !!tool.returns,
hasAiAgent: !!tool.aiAgent,
});
await prisma.tool.update({
where: { id: tool.id },
data: {
qualityScore,
},
});
}
return pkg.npmPackageName;
})
);
for (const result of results) {
if (result.status === 'fulfilled') {
processed++;
} else {
errors++;
const errorMsg = `Failed to process package: ${result.reason instanceof Error ? result.reason.message : 'Unknown error'}`;
errorMessages.push(errorMsg);
console.error(errorMsg);
}
}
}
// Update checkpoint with last run timestamp
await prisma.syncCheckpoint.upsert({
where: { source: 'metrics' },
create: {
@ -106,7 +114,6 @@ export async function POST(request: NextRequest) {
},
});
// Log sync operation
await prisma.syncLog.create({
data: {
source: 'metrics',
@ -140,7 +147,6 @@ export async function POST(request: NextRequest) {
} catch (error) {
console.error('Metrics sync failed:', error);
// Log failed sync
await prisma.syncLog.create({
data: {
source: 'metrics',

4
vercel.json
View file

@ -49,6 +49,10 @@
"path": "/api/sync/keyword",
"schedule": "0 */6 * * *"
},
{
"path": "/api/sync/enrich",
"schedule": "*/2 * * * *"
},
{
"path": "/api/sync/metrics",
"schedule": "0 0 * * *"