app/scripts/smoke-worker.mjs
Rama main · réplica de lectura del repositorio original.
import assert from 'node:assert/strict'
import { spawn, spawnSync } from 'node:child_process'
import {
mkdtempSync,
mkdirSync,
readFileSync,
realpathSync,
writeFileSync,
} from 'node:fs'
import { createServer } from 'node:net'
import { resolve } from 'node:path'
import { fileURLToPath } from 'node:url'
import { setTimeout as delay } from 'node:timers/promises'
import { parse } from 'jsonc-parser'
import { createClient } from '@libsql/client/web'
import { smokeMcp } from './smoke-mcp.mjs'
import { smokeAssets } from './smoke-assets.mjs'
import { smokePlugins } from './smoke-plugins.mjs'
import { smokePluginServices } from './smoke-plugin-services.mjs'
const appDir = resolve(
process.env.DEVPUSH_SMOKE_APP_DIR ||
fileURLToPath(new URL('../', import.meta.url)),
)
const consumer = Boolean(process.env.DEVPUSH_SMOKE_APP_DIR)
const packageDir = consumer
? realpathSync(resolve(appDir, 'node_modules/devpush'))
: appDir
const development = process.env.DEVPUSH_SMOKE_DEVELOPMENT === '1'
if (development && !consumer)
throw new Error(
'Worker development smoke requires a disposable package consumer.',
)
const driver = process.argv[2]
if (!['d1', 'libsql'].includes(driver)) throw new Error('Choose d1 or libsql.')
mkdirSync(resolve(appDir, '.tmp'), { recursive: true })
const directory = mkdtempSync(resolve(appDir, '.tmp/worker-smoke-'))
const configPath = resolve(directory, 'wrangler.jsonc')
const secretsPath = resolve(directory, 'secrets.json')
const statePath = development
? resolve(appDir, '.wrangler/state')
: resolve(directory, 'state')
const listener = createServer()
await new Promise((resolve) => listener.listen(0, '127.0.0.1', resolve))
const { port } = listener.address()
await new Promise((resolve, reject) =>
listener.close((error) => (error ? reject(error) : resolve())),
)
const origin = `http://127.0.0.1:${port}`
const config = parse(
readFileSync(resolve(packageDir, 'wrangler.template.jsonc'), 'utf8'),
)
config.main = resolve(packageDir, 'src/server.ts')
config.vars = {
DEVPUSH_BASE_URL: origin,
BETTER_AUTH_URL: origin,
DEVPUSH_MCP_URL: `${origin}/mcp`,
BETTER_AUTH_SECRET: 'worker-smoke-auth-secret-over-thirty-two-characters',
EMAIL_FROM: 'Smoke <smoke@example.invalid>',
RESEND_API_KEY: 're_worker_smoke_invalid_key',
DEVPUSH_SECRET_ENCRYPTION_KEY:
'worker-smoke-encryption-key-over-thirty-two-characters',
}
delete config.r2_buckets // Prove that a Worker can run without image storage.
const clients = []
if (driver === 'libsql') {
delete config.d1_databases
for (const [key, fallback] of [
['DEVPUSH_DATABASE_URL', 'http://127.0.0.1:7561'],
['DEVPUSH_LOG_DATABASE_URL', 'http://127.0.0.1:7562'],
]) {
const url = process.env[key] || fallback
if (!['127.0.0.1', 'localhost', '[::1]'].includes(new URL(url).hostname))
throw new Error(
'Worker smoke tests require disposable loopback libSQL servers.',
)
config.vars[key] = url
clients.push(createClient({ url }))
}
} else {
for (const binding of config.d1_databases)
binding.migrations_dir = resolve(packageDir, binding.migrations_dir)
}
writeFileSync(configPath, JSON.stringify(config))
writeFileSync(secretsPath, '{}')
const env = {
...process.env,
DEVPUSH_DEPLOY_TARGET: 'cloudflare',
DEVPUSH_WRANGLER_CONFIG: configPath,
DEVPUSH_CLOUDFLARE_SECRETS_FILE: secretsPath,
}
function run(command, args, capture = false) {
const result = spawnSync(command, args, {
cwd: appDir,
env,
stdio: capture ? 'pipe' : 'inherit',
encoding: 'utf8',
})
if (result.error) throw result.error
if (result.status !== 0)
throw new Error(
`${command} ${args.join(' ')} failed: ${result.stderr || result.status}`,
)
return result.stdout
}
function capture(command, args) {
return new Promise((resolve, reject) => {
const child = spawn(command, args, {
cwd: appDir,
env,
stdio: ['ignore', 'pipe', 'pipe'],
})
let stdout = ''
let stderr = ''
child.stdout.on('data', (chunk) => {
stdout += chunk
})
child.stderr.on('data', (chunk) => {
stderr += chunk
})
child.once('error', reject)
child.once('close', (code) => {
if (code === 0) resolve(stdout)
else reject(new Error(`D1 fixture command failed (${code}): ${stderr}`))
})
})
}
async function d1(index, sql) {
// Keep the live HTTP client's timers, socket events and Wrangler output
// processing responsive while fixture commands run.
const output = await capture(process.execPath, [
'node_modules/wrangler/bin/wrangler.js',
'd1',
'execute',
index === 0 ? 'DEVPUSH_DATABASE' : 'DEVPUSH_LOG_DATABASE',
'--local',
'--config',
configPath,
'--persist-to',
statePath,
'--command',
sql,
'--json',
])
return JSON.parse(output)[0].results
}
async function query(index, sql) {
return driver === 'd1'
? d1(index, sql)
: (await clients[index].execute(sql)).rows
}
let worker
let output = ''
try {
for (const client of clients) {
for (let attempt = 0; ; attempt++) {
try {
await client.execute('select 1')
break
} catch (error) {
if (attempt === 39) throw error
await delay(250)
}
}
}
if (consumer) {
assert.equal(driver, 'd1', 'Package proof uses disposable local D1')
for (const binding of config.d1_databases) {
run('pnpm', [
'exec',
'wrangler',
'd1',
'migrations',
'apply',
binding.binding,
'--config',
configPath,
'--local',
'--persist-to',
statePath,
])
}
if (!development) run('pnpm', ['exec', 'vite', 'build'])
} else {
run('pnpm', [
'db:migrate:cloudflare',
...(driver === 'd1' ? ['--local', '--persist-to', statePath] : []),
])
run('pnpm', ['build:cloudflare'])
}
await query(
0,
"insert into rate_limits(key,count,reset_at) values('worker-smoke-expired',1,0)",
)
await query(
1,
"insert into runtime_logs(id,project_id,environment_id,server_id,runtime_id,service,observed_at,content,search_content,content_hash) values('worker-smoke-expired','p','e','s','r','web',0,'old','old','worker-smoke-hash')",
)
worker = spawn(
process.execPath,
development
? [
'node_modules/vite/bin/vite.js',
'dev',
'--host',
'127.0.0.1',
'--port',
String(port),
'--strictPort',
]
: [
'node_modules/wrangler/bin/wrangler.js',
'dev',
'--config',
'dist/server/wrangler.json',
'--persist-to',
statePath,
'--port',
String(port),
'--test-scheduled',
],
{
cwd: appDir,
env,
detached: process.platform !== 'win32',
stdio: ['ignore', 'pipe', 'pipe'],
},
)
worker.stdout.on('data', (chunk) => {
output += chunk
})
worker.stderr.on('data', (chunk) => {
output += chunk
})
const startupDeadline = Date.now() + 120_000
for (;;) {
try {
const response = await fetch(`${origin}/api/health`, {
// Cold source-package transforms can outlast a production health probe.
signal: AbortSignal.timeout(
Math.min(
development ? 30_000 : 3000,
Math.max(1, startupDeadline - Date.now()),
),
),
})
assert.equal(response.status, 200)
break
} catch (error) {
if (
Date.now() >= startupDeadline ||
worker.exitCode !== null ||
worker.signalCode !== null
)
throw error
await delay(250)
}
}
const login = await fetch(`${origin}/auth/login`)
assert.equal(login.status, 200)
await smokeAssets(origin, await login.text(), { production: !development })
await smokePlugins(origin)
// Authenticated routes render a public skeleton without waiting for session/DB
// loaders; client server-functions still enforce authorization when hydrated.
for (const path of [
'/smoke-team',
'/smoke-team/projects/demo/settings',
'/settings',
]) {
const response = await fetch(`${origin}${path}`, { redirect: 'manual' })
assert.equal(
response.status,
200,
`Expected an immediate pending document for ${path}`,
)
const html = await response.text()
assert.match(
html,
/data-slot="skeleton"/,
`Missing pending layout for ${path}`,
)
}
const adminResponse = await fetch(`${origin}/admin`, { redirect: 'manual' })
assert.equal(adminResponse.status, 307)
const adminLocation = new URL(adminResponse.headers.get('location'), origin)
assert.equal(adminLocation.origin, origin)
assert.equal(adminLocation.pathname, '/admin')
const canonicalAdmin = await fetch(adminLocation)
assert.equal(canonicalAdmin.status, 200)
assert.match(await canonicalAdmin.text(), /data-slot="skeleton"/)
// Exercise auth writes without sending an email: rate limiting runs before
// Better Auth rejects the malformed address.
const invalidEmail = await fetch(
`${origin}/api/auth/email-otp/send-verification-otp`,
{
method: 'POST',
headers: { 'content-type': 'application/json', origin },
body: JSON.stringify({ email: 'invalid-email', type: 'sign-in' }),
},
)
assert.equal(invalidEmail.status, 400, await invalidEmail.text())
await smokeMcp({
origin,
query,
appDir,
development,
modulePrefix: development ? `@fs/${packageDir}/src` : 'src',
authSecret: config.vars.BETTER_AUTH_SECRET,
})
await smokePluginServices({
origin,
authSecret: config.vars.BETTER_AUTH_SECRET,
query,
appDir,
development,
modulePrefix: development ? `@fs/${packageDir}/src` : 'src',
bundleDir: 'dist/server/assets',
})
if (development) {
console.log(
'Worker Vite development: SSR, styles, D1, WebSockets and API/MCP authorization passed.',
)
} else {
await query(
0,
"insert into teams(id,name,slug,status,updated_at) values('worker-smoke-deleted','Deleted team','worker-smoke-deleted','deleted',0)",
)
const responses = await Promise.all(
Array.from({ length: 20 }, () => fetch(`${origin}/api/health`)),
)
assert(responses.every((response) => response.status === 200))
await Promise.all(responses.map((response) => response.arrayBuffer()))
const time = Math.floor(Date.now() / 3_600_000) * 3_600_000
assert.equal(
(
await fetch(
`${origin}/cdn-cgi/local/scheduled?cron=0+*+*+*+*&time=${time}`,
)
).status,
200,
)
for (let attempt = 0; ; attempt++) {
const main = await query(
0,
"select count(*) as n from rate_limits where key='worker-smoke-expired'",
)
const logs = await query(
1,
"select count(*) as n from runtime_logs where id='worker-smoke-expired'",
)
const deleted = await query(
0,
"select count(*) as n from teams where id='worker-smoke-deleted'",
)
if (
main[0].n === 0 &&
logs[0].n === 0 &&
deleted[0].n === 0
)
break
if (attempt === 19)
throw new Error(
'Scheduled cleanup did not finish.',
)
await delay(250)
}
console.log(
`Worker ${driver}: login, API/MCP browser authorization, concurrent health requests and scheduled resource/log cleanup passed.`,
)
}
} catch (error) {
console.error(output)
throw error
} finally {
writeFileSync(resolve(directory, 'worker.log'), output)
if (worker && worker.exitCode === null && worker.signalCode === null) {
if (process.platform === 'win32') worker.kill('SIGTERM')
else process.kill(-worker.pid, 'SIGTERM')
await new Promise((resolve) => worker.once('exit', resolve))
}
clients.forEach((client) => client.close())
}