chore: kanban template
This commit is contained in:
e2e
203
packages/functions/scripts/local-server.ts
Normal file
203
packages/functions/scripts/local-server.ts
Normal file
@@ -0,0 +1,203 @@
|
||||
// Local single-process harness for end-to-end testing the kanban workflow
|
||||
// path WITHOUT Cloudflare. Hosts the kanban HTTP wirings, fakes the fabric
|
||||
// `/tenant/enqueue` endpoint, and runs queue jobs in-process via the same
|
||||
// `runQueueJob` the CF handler-factories use. Workflow services are wired
|
||||
// the same way the generated CF entry does (SQLiteKyselyWorkflowService +
|
||||
// SQLiteKyselyWorkflowRunService over libsql).
|
||||
//
|
||||
// Usage:
|
||||
// export DATABASE_URL='libsql://...?authToken=...' # MUST be set
|
||||
// npx tsx scripts/local-server.ts
|
||||
//
|
||||
// Then:
|
||||
// curl -X POST http://127.0.0.1:9100/workflow/cardOnboardingWorkflow/start \
|
||||
// -H 'content-type: application/json' \
|
||||
// -d '{"data":{"title":"local-e2e","status":"todo"}}'
|
||||
// # poll /workflow/cardOnboardingWorkflow/status/<runId>
|
||||
|
||||
import http from 'node:http'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { Kysely, CamelCasePlugin } from 'kysely'
|
||||
import {
|
||||
LibsqlWebDialect,
|
||||
SQLiteKyselyWorkflowService,
|
||||
SQLiteKyselyWorkflowRunService,
|
||||
type KyselyPikkuDB,
|
||||
} from '@pikku/kysely-sqlite'
|
||||
import { SerializePlugin } from '@pikku/kysely'
|
||||
import { runQueueJob, type QueueJob, type QueueJobStatus } from '@pikku/core/queue'
|
||||
import { fetchData, PikkuFetchHTTPResponse } from '@pikku/core/http'
|
||||
import { compileAllSchemas } from '@pikku/core/schema'
|
||||
import { incomingMessageToRequest, writeResponse } from '@pikku/node-http-server'
|
||||
import { createConfig } from '../src/config.js'
|
||||
import { createSingletonServices } from '../src/services.js'
|
||||
import '../.pikku/pikku-bootstrap.gen.js'
|
||||
|
||||
const PORT = Number(process.env.PORT ?? 9100)
|
||||
const HOST = process.env.HOST ?? '127.0.0.1'
|
||||
const STAGE_ID = process.env.FABRIC_STAGE_ID ?? 'local-stage'
|
||||
const DATABASE_URL = process.env.DATABASE_URL
|
||||
if (!DATABASE_URL) {
|
||||
console.error(
|
||||
'DATABASE_URL required (libsql URL with embedded authToken). Pull one from app.stage.database_url.',
|
||||
)
|
||||
process.exit(1)
|
||||
}
|
||||
|
||||
// Mirrors the inline FabricTenantQueueService used by the deploy runtime.
|
||||
// Same body shape, no HMAC since the local /tenant/enqueue endpoint accepts
|
||||
// any caller.
|
||||
class FabricTenantQueueServiceLocal {
|
||||
supportsResults = false
|
||||
constructor(
|
||||
private readonly endpoint: string,
|
||||
private readonly stageId: string,
|
||||
) {}
|
||||
async add(queueName: string, data: unknown, options?: { jobId?: string }): Promise<string> {
|
||||
const body = JSON.stringify({
|
||||
stageId: this.stageId,
|
||||
pikkuQueueName: queueName,
|
||||
payload: data,
|
||||
jobId: options?.jobId,
|
||||
})
|
||||
const res = await fetch(this.endpoint, {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body,
|
||||
})
|
||||
if (!res.ok) {
|
||||
const text = await res.text().catch(() => '')
|
||||
throw new Error(
|
||||
`local tenant-queue: enqueue ${queueName} → ${res.status} ${text.slice(0, 200)}`,
|
||||
)
|
||||
}
|
||||
const json = (await res.json()) as { jobId: string }
|
||||
return json.jobId
|
||||
}
|
||||
async getJob(): Promise<null> {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function mkJob(queueName: string, data: unknown, id: string): QueueJob {
|
||||
return {
|
||||
queueName,
|
||||
id,
|
||||
data,
|
||||
status: async () => 'active' as QueueJobStatus,
|
||||
metadata: () => ({
|
||||
processedAt: new Date(),
|
||||
attemptsMade: 0,
|
||||
maxAttempts: undefined,
|
||||
result: undefined,
|
||||
progress: 0,
|
||||
createdAt: new Date(),
|
||||
completedAt: undefined,
|
||||
failedAt: undefined,
|
||||
error: undefined,
|
||||
}),
|
||||
waitForCompletion: async () => {
|
||||
throw new Error('local: waitForCompletion not supported')
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
async function readBody(req: http.IncomingMessage): Promise<string> {
|
||||
const chunks: Buffer[] = []
|
||||
for await (const c of req as AsyncIterable<Buffer>) chunks.push(c)
|
||||
return Buffer.concat(chunks).toString('utf-8')
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const config = await createConfig()
|
||||
|
||||
const queueEndpoint = `http://${HOST}:${PORT}/tenant/enqueue`
|
||||
const queueService = new FabricTenantQueueServiceLocal(queueEndpoint, STAGE_ID)
|
||||
|
||||
// Pass queueService through existingServices — the template's
|
||||
// createSingletonServices preserves anything in existingServices that
|
||||
// it doesn't explicitly override.
|
||||
const services = (await createSingletonServices(config, {
|
||||
queueService: queueService as never,
|
||||
})) as Record<string, unknown>
|
||||
|
||||
const workflowKysely = new Kysely<KyselyPikkuDB>({
|
||||
dialect: new LibsqlWebDialect({ url: DATABASE_URL! }),
|
||||
plugins: [new CamelCasePlugin(), new SerializePlugin()],
|
||||
})
|
||||
const workflowService = new SQLiteKyselyWorkflowService(workflowKysely)
|
||||
await workflowService.init()
|
||||
services.workflowService = workflowService
|
||||
services.workflowRunService = new SQLiteKyselyWorkflowRunService(workflowKysely)
|
||||
|
||||
compileAllSchemas((services as { logger: { info: (s: string) => void } }).logger as never)
|
||||
|
||||
const log = (services as { logger: { info: (s: string) => void; error: (s: string) => void } })
|
||||
.logger
|
||||
|
||||
const server = http.createServer(async (req, res) => {
|
||||
try {
|
||||
if (req.method === 'POST' && req.url === '/tenant/enqueue') {
|
||||
const text = await readBody(req)
|
||||
let body: {
|
||||
stageId?: string
|
||||
pikkuQueueName: string
|
||||
payload: unknown
|
||||
jobId?: string
|
||||
}
|
||||
try {
|
||||
body = JSON.parse(text)
|
||||
} catch {
|
||||
res.writeHead(400, { 'content-type': 'application/json' })
|
||||
res.end(JSON.stringify({ error: 'invalid_json' }))
|
||||
return
|
||||
}
|
||||
const jobId = body.jobId ?? randomUUID()
|
||||
log.info(`[enqueue] queue=${body.pikkuQueueName} jobId=${jobId} stage=${body.stageId}`)
|
||||
// Async, fire-and-forget — mirrors pg-boss behaviour: the producer
|
||||
// gets an immediate ack and the job runs on the worker side.
|
||||
void (async () => {
|
||||
try {
|
||||
await runQueueJob({
|
||||
job: mkJob(body.pikkuQueueName, body.payload, jobId),
|
||||
})
|
||||
log.info(`[queue:${body.pikkuQueueName}] ${jobId} done`)
|
||||
} catch (e) {
|
||||
const err = e as Error
|
||||
log.error(
|
||||
`[queue:${body.pikkuQueueName}] ${jobId} FAILED: ${err.message}\n${err.stack ?? ''}`,
|
||||
)
|
||||
}
|
||||
})()
|
||||
res.writeHead(200, { 'content-type': 'application/json' })
|
||||
res.end(JSON.stringify({ jobId }))
|
||||
return
|
||||
}
|
||||
|
||||
const request = incomingMessageToRequest(req)
|
||||
const pikkuResponse = new PikkuFetchHTTPResponse()
|
||||
await fetchData(request, pikkuResponse, { respondWith404: true })
|
||||
await writeResponse(res, pikkuResponse.toResponse())
|
||||
} catch (err) {
|
||||
log.error(`local-server: ${(err as Error).message}`)
|
||||
if (!res.headersSent) {
|
||||
res.writeHead(500, { 'content-type': 'application/json' })
|
||||
}
|
||||
try {
|
||||
res.end(JSON.stringify({ error: 'internal_error' }))
|
||||
} catch {
|
||||
// already ended
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
server.listen(PORT, HOST, () => {
|
||||
log.info(`local-server: listening on http://${HOST}:${PORT}`)
|
||||
log.info(`local-server: stage=${STAGE_ID} db=${DATABASE_URL!.split('?')[0]}`)
|
||||
})
|
||||
}
|
||||
|
||||
main().catch((e) => {
|
||||
console.error('local-server: fatal', e)
|
||||
process.exit(1)
|
||||
})
|
||||
Reference in New Issue
Block a user