cli-e2e-mqo21uyk/packages/functions/scripts/local-server.ts

// 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)
})