server.ts

import express from "express";
import path from "path";

import { mysql, redis } from "./db/client";
import { startWorker } from "./worker/dispatcher";
import subscriptionsRouter from "./routes/subscriptions";
import eventsRouter from "./routes/events";
import dlqRouter from "./routes/dlq";
import jobsRouter from "./routes/jobs";
import { log } from "./lib/logger";

const app = express();
const port = process.env.PORT || 8787;

export { mysql, redis };

app.use(express.json());
app.use(express.static(path.join(__dirname, "public")));

app.get("/health", async (req, res) => {
  const [mysqlOk, redisOk] = await Promise.allSettled([
    mysql`SELECT 1`,
    redis.ping(),
  ]);
  const [{ count }] = await mysql`
    SELECT COUNT(*) as count FROM delivery_jobs WHERE status = 'pending'
  `;
  res.json({
    status: "ok",
    mysql: mysqlOk.status === "fulfilled" ? "connected" : "error",
    redis: redisOk.status === "fulfilled" ? "connected" : "error",
    queue_depth: Number(count),
  });
});

app.use("/subscriptions", subscriptionsRouter);
app.use("/events", eventsRouter);
app.use("/dlq", dlqRouter);
app.use("/jobs", jobsRouter);

app.get("/api/test/mysql", async (req, res) => {
  const rows = await mysql`SELECT 'test'`.values();
  res.json({ data: rows });
});

app.get("/api/test/redis", async (req, res) => {
  const pingresp = await redis.ping();
  res.json({ data: pingresp });
});

async function migrate() {
  const schema = await Bun.file("db/schema.sql").text();
  for (const statement of schema.split(";").filter(s => s.trim())) {
    await mysql.unsafe(statement);
  }
}

async function main() {
  log("info", "server.migrating");
  await migrate();
  log("info", "server.listening", { port });

  startWorker(Number(process.env.WORKER_CONCURRENCY) || 5);
}

app.listen(port, main);