Files
triggerdotdev--trigger.dev/apps/webapp/server.ts
T
Eric Allam 2d8a41b18b feat: realtime (#1402)
* Denormalize run tags, increase character limit to 128

* WIP realtime subscribing to runs

* extracted the stream stuff into core, made it more reusable

* WIP tags

* Remove tags for now because it’s not support in electric

* Support async iterables, readable stream, and callback style subscription styles

* Remove tags streaming endpoint

* Add realtime rate limits and scope them to the /realtime path

* WIP rate limt per org

* Introduce per org rate limits

* WIP JWT auth

* Move migrations into new internal db package

* Resolve pnpm lock file

* Authenticating to the realtime API with JWTs are working

* realtime in the client

* Created react-hooks package and starting to move stuff in there

* Improve types for hooks

* schema tasks

* Added useBatch hook

* build uploadthing/fal demo and change how run metadata is synced to the server

* tweaks

* WIL realtime concurrency tracking

* Implement test for realtime client using testcontainers

also updated electric to latest version

* Allow customizing the expiration time of the automatic JWT created after triggering a task

* Add support for subscribing to run tags

* Improve auth types and API

* finalize the realtime API

* Fixed some example stuff

* Allow up to 10 run tags

* Remove core from docker-provider tsconfig paths to prevent it from being typechecked

* do the same for the kubernetes provider

* Fixing some typecheck errors

* Fix webapp type errors

* Update @trigger.dev/platform to 1.0.13

* Fix attw error

* Remove from/to in subscribeToRuns query params

* Add tests for the rate limit middleware and add custom JWT rate limits

* turn off webapp test parallelism

* Finish renaming jwt -> publicAccessToken and automatically give the JWT read access to the tags when using trigger

* Add changeset

* Attempt to fix unit tests in CI

* Skip running the auth rate limit middleware tests for now

* Try a beefier machine

* Try and run webapp tests separately

* Setup env vars

* Make sliding window test more reliabile
2024-10-21 15:07:08 +01:00

170 lines
5.4 KiB
TypeScript

import { createRequestHandler } from "@remix-run/express";
import { broadcastDevReady, logDevReady } from "@remix-run/server-runtime";
import compression from "compression";
import type { Server as EngineServer } from "engine.io";
import express from "express";
import morgan from "morgan";
import { nanoid } from "nanoid";
import path from "path";
import type { Server as IoServer } from "socket.io";
import { WebSocketServer } from "ws";
import { RateLimitMiddleware } from "~/services/apiRateLimit.server";
import { type RunWithHttpContextFunction } from "~/services/httpAsyncStorage.server";
import { RegistryProxy } from "~/v3/registryProxy.server";
const app = express();
if (process.env.DISABLE_COMPRESSION !== "1") {
app.use(compression());
}
// http://expressjs.com/en/advanced/best-practice-security.html#at-a-minimum-disable-x-powered-by-header
app.disable("x-powered-by");
// Remix fingerprints its assets so we can cache forever.
app.use("/build", express.static("public/build", { immutable: true, maxAge: "1y" }));
// Everything else (like favicon.ico) is cached for an hour. You may want to be
// more aggressive with this caching.
app.use(express.static("public", { maxAge: "1h" }));
app.use(morgan("tiny"));
process.title = "node webapp-server";
const MODE = process.env.NODE_ENV;
const BUILD_DIR = path.join(process.cwd(), "build");
const build = require(BUILD_DIR);
const port = process.env.REMIX_APP_PORT || process.env.PORT || 3000;
if (process.env.HTTP_SERVER_DISABLED !== "true") {
const socketIo: { io: IoServer } | undefined = build.entry.module.socketIo;
const wss: WebSocketServer | undefined = build.entry.module.wss;
const registryProxy: RegistryProxy | undefined = build.entry.module.registryProxy;
const apiRateLimiter: RateLimitMiddleware = build.entry.module.apiRateLimiter;
const runWithHttpContext: RunWithHttpContextFunction = build.entry.module.runWithHttpContext;
if (registryProxy && process.env.ENABLE_REGISTRY_PROXY === "true") {
console.log(`🐳 Enabling container registry proxy to ${registryProxy.origin}`);
// Adjusted to match /v2 and any subpath under /v2
app.all("/v2/*", async (req, res) => {
await registryProxy.call(req, res);
});
// This might also be necessary if you need to explicitly match /v2 as well
app.all("/v2", async (req, res) => {
await registryProxy.call(req, res);
});
}
app.use((req, res, next) => {
// helpful headers:
res.set("Strict-Transport-Security", `max-age=${60 * 60 * 24 * 365 * 100}`);
// /clean-urls/ -> /clean-urls
if (req.path.endsWith("/") && req.path.length > 1) {
const query = req.url.slice(req.path.length);
const safepath = req.path.slice(0, -1).replace(/\/+/g, "/");
res.redirect(301, safepath + query);
return;
}
next();
});
app.use((req, res, next) => {
// Generate a unique request ID for each request
const requestId = nanoid();
runWithHttpContext({ requestId, path: req.url, host: req.hostname }, next);
});
if (process.env.DASHBOARD_AND_API_DISABLED !== "true") {
app.use(apiRateLimiter);
app.all(
"*",
// @ts-ignore
createRequestHandler({
build,
mode: MODE,
})
);
} else {
// we need to do the health check here at /healthcheck
app.get("/healthcheck", (req, res) => {
res.status(200).send("OK");
});
}
const server = app.listen(port, () => {
console.log(`✅ server ready: http://localhost:${port} [NODE_ENV: ${MODE}]`);
if (MODE === "development") {
broadcastDevReady(build)
.then(() => logDevReady(build))
.catch(console.error);
}
});
server.keepAliveTimeout = 65 * 1000;
process.on("SIGTERM", () => {
server.close((err) => {
if (err) {
console.error("Error closing express server:", err);
} else {
console.log("Express server closed gracefully.");
}
});
});
socketIo?.io.attach(server);
server.removeAllListeners("upgrade"); // prevent duplicate upgrades from listeners created by io.attach()
server.on("upgrade", async (req, socket, head) => {
console.log(
`Attemping to upgrade connection at url ${req.url} with headers: ${JSON.stringify(
req.headers
)}`
);
socket.on("error", (err) => {
console.error("Connection upgrade error:", err);
});
const url = new URL(req.url ?? "", "http://localhost");
// Upgrade socket.io connection
if (url.pathname.startsWith("/socket.io/")) {
console.log(`Socket.io client connected, upgrading their connection...`);
// https://github.com/socketio/socket.io/issues/4693
(socketIo?.io.engine as EngineServer).handleUpgrade(req, socket, head);
return;
}
// Only upgrade the connecting if the path is `/ws`
if (url.pathname !== "/ws") {
// Setting the socket.destroy() error param causes an error event to be emitted which needs to be handled with socket.on("error") to prevent uncaught exceptions.
socket.destroy(
new Error(
"Cannot connect because of invalid path: Please include `/ws` in the path of your upgrade request."
)
);
return;
}
console.log(`Client connected, upgrading their connection...`);
// Handle the WebSocket connection
wss?.handleUpgrade(req, socket, head, (ws) => {
wss?.emit("connection", ws, req);
});
});
} else {
require(BUILD_DIR);
console.log(`✅ app ready (skipping http server)`);
}