Connect to a private GCP resource from a deployed task

import { spawn } from "node:child_process";
import net from "node:net";
import { setTimeout as sleep } from "node:timers/promises";
import { SocksClient } from "socks";

const STATE_DIR = process.env.TS_STATE_DIR ?? "/home/node/.ts-state";
const SOCKS_HOST = "127.0.0.1";
const SOCKS_PORT = Number(process.env.TS_SOCKS_PORT ?? 1055);
const SOCKET = `${STATE_DIR}/tailscaled.sock`;

let readyPromise: Promise<void> | undefined;

/** Idempotently start userspace tailscaled and resolve once the tailnet is up. */
export function ensureTunnel(): Promise<void> {
  if (readyPromise) return readyPromise;
  readyPromise = start().catch((err) => {
    readyPromise = undefined;
    throw err;
  });
  return readyPromise;
}

async function start(): Promise<void> {
  const authKey = process.env.TS_AUTHKEY;
  if (!authKey) throw new Error("TS_AUTHKEY is not set");

  // Userspace networking means no TUN device and no NET_ADMIN, so it runs as the task user.
  const daemon = spawn(
    "tailscaled",
    [
      "--tun=userspace-networking",
      `--socks5-server=${SOCKS_HOST}:${SOCKS_PORT}`,
      `--state=${STATE_DIR}/tailscaled.state`,
      `--socket=${SOCKET}`,
    ],
    { stdio: "inherit" }
  );
  daemon.on("exit", (code) => console.error(`[tunnel] tailscaled exited: ${code}`));

  // Wait for the daemon to be listening before authenticating, or `tailscale up`
  // races the cold daemon, exits, and the backend wait below times out.
  await waitForPort(SOCKS_HOST, SOCKS_PORT, 15_000);

  await run("tailscale", [
    `--socket=${SOCKET}`,
    "up",
    "--timeout=30s", // bound the handshake so a failed auth fails fast instead of hanging the run
    `--authkey=${authKey}`,
    `--hostname=${process.env.TS_HOSTNAME ?? "trigger-task"}`,
    "--accept-routes",
    // Point at a self-hosted Headscale by setting TS_LOGIN_SERVER (see below).
    ...(process.env.TS_LOGIN_SERVER ? [`--login-server=${process.env.TS_LOGIN_SERVER}`] : []),
  ]);

  await waitForBackendRunning(30_000);
}

/** Open a TCP socket to a private host:port through the SOCKS5 proxy. */
export async function tunnelSocket(host: string, port: number): Promise<net.Socket> {
  const info = await SocksClient.createConnection({
    proxy: { host: SOCKS_HOST, port: SOCKS_PORT, type: 5 },
    command: "connect",
    destination: { host, port },
    timeout: 10_000,
  });
  return info.socket;
}

async function backendState(): Promise<string | null> {
  const out = await run("tailscale", [`--socket=${SOCKET}`, "status", "--json"], true).catch(
    () => ""
  );
  return /"BackendState"\s*:\s*"([^"]*)"/.exec(out)?.[1] ?? null;
}

async function waitForBackendRunning(timeoutMs: number): Promise<void> {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if ((await backendState()) === "Running") return;
    await sleep(500);
  }
  throw new Error("[tunnel] tailnet did not reach Running");
}

function waitForPort(host: string, port: number, timeoutMs: number): Promise<void> {
  const deadline = Date.now() + timeoutMs;
  return (async () => {
    while (Date.now() < deadline) {
      const open = await new Promise<boolean>((resolve) => {
        const s = net.connect(port, host, () => {
          s.destroy();
          resolve(true);
        });
        s.on("error", () => resolve(false));
      });
      if (open) return;
      await sleep(300);
    }
    throw new Error(`[tunnel] proxy port ${host}:${port} never opened`);
  })();
}

/** Run a child process, capturing stdout when `capture` is set. */
function run(cmd: string, args: string[], capture = false): Promise<string> {
  return new Promise((resolve, reject) => {
    const p = spawn(cmd, args, { stdio: capture ? ["ignore", "pipe", "pipe"] : "inherit" });
    let out = "";
    if (capture) p.stdout?.on("data", (d) => (out += d));
    p.on("exit", (code) =>
      code === 0 || capture ? resolve(out) : reject(new Error(`${cmd} exited ${code}`))
    );
    p.on("error", reject);
  });
}

import { defineConfig, tasks } from "@trigger.dev/sdk";
import { tailscaleTunnel } from "./extension/tailscaleTunnel.js";
import { ensureTunnel } from "./runtime/tunnel.js";

// Start eagerly at worker boot, but never during the build step.
const isBuild = !!process.env.TRIGGER_BUILD_MANIFEST_PATH;
if (process.env.TS_AUTHKEY && !isBuild) void ensureTunnel().catch(() => {});

tasks.middleware("tunnel", async ({ next }) => {
  await ensureTunnel();
  await next();
});

export default defineConfig({
  project: "<your-project-ref>",
  maxDuration: 300,
  dirs: ["./src"],
  processKeepAlive: true, // reuse the worker process, and therefore the tunnel, across runs
  build: {
    extensions: [tailscaleTunnel({ arch: "amd64" })],
    external: ["tailscale", "tailscaled"],
  },
});

import pg from "pg";
import { tunnelSocket } from "./tunnel.js";

/** Run a single query against the private Postgres through the tunnel. */
export async function pgSelect(sql: string): Promise<Record<string, unknown>[]> {
  const client = new pg.Client({
    user: process.env.PGUSER,
    password: process.env.PGPASSWORD,
    database: process.env.PGDATABASE,
    // The socket is already connected through the SOCKS5 proxy.
    stream: await tunnelSocket(process.env.PGHOST!, Number(process.env.PGPORT ?? 5432)),
  });
  await client.connect();
  try {
    const { rows } = await client.query(sql);
    return rows;
  } finally {
    await client.end();
  }
}

import { task } from "@trigger.dev/sdk";
import { pgSelect } from "../runtime/db.js";

export const queryPrivateGcp = task({
  id: "query-private-gcp",
  run: async () => {
    // inet_server_addr() returns the private server IP, proving the query traversed the tunnel.
    const rows = await pgSelect("select now() as ts, inet_server_addr() as ip, version()");
    return { postgres: rows[0] };
  },
});

TS_LOGIN_SERVER=https://headscale.your-domain.example
TS_AUTHKEY=<a Headscale pre-auth key>