Runner: Proxy und Startseite, register/wake (R12)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

runner/src/dev.js

@@ -42,6 +42,7 @@ export function createDev({
   setTimer = (fn, ms) => setTimeout(fn, ms).unref?.() ?? 0, prepareDirs = defaultPrepareDirs,
   rmDir = (p) => nodeFs.rm(p, { recursive: true, force: true }),
 }) {
+  const specs = new Map(); // R12: letzte Job-Spezifikation je sub (bleibt auch im Zustand 'sleeping')
   const entries = new Map(); // sub ⇒ { sub, state, sha, lastUsed, log, port, job, release, promise, cancelled }
   const sleep = (ms) => new Promise((r) => setTimer(r, ms));
   const isRunning = (e) => e.state === 'online' || e.state === 'starting';
@@ -123,6 +124,7 @@ export function createDev({
   }
 
   async function ensure(sub, job) {
+    specs.set(sub, job);
     const cur = entries.get(sub);
     if (cur && cur.sha === job.sha) {
       if (cur.state === 'failed' && now() - cur.failedAt < RETRY_MS) return { state: 'failed' }; // nicht öfter als alle 60 s neu versuchen
@@ -142,8 +144,24 @@ export function createDev({
     return { state: 'starting' };
   }
 
+  // R12: Spezifikation merken, ohne zu starten. Läuft der sub mit anderem sha, wird nach den üblichen Regeln neu gestartet.
+  async function register(sub, job) {
+    const cur = entries.get(sub);
+    if (cur && isRunning(cur) && cur.sha !== job.sha) return ensure(sub, job);
+    specs.set(sub, job);
+    if (cur && cur.sha === job.sha && (cur.state === 'online' || cur.state === 'starting' || cur.state === 'failed')) return { state: cur.state };
+    return { state: 'sleeping' };
+  }
+  // R12: Start mit der gemerkten Spezifikation (Proxy kennt nur die Subdomain).
+  async function wake(sub) {
+    const job = specs.get(sub);
+    if (!job) return { state: 'unknown' };
+    return ensure(sub, job);
+  }
+
   const touch = (sub) => { const e = entries.get(sub); if (e) e.lastUsed = now(); };
   async function stop(sub) {
+    specs.delete(sub); // 'off': Spezifikation vergessen, bis das Portal neu registriert
     const e = entries.get(sub);
     if (!e) return;
     await shutdown(e, 'off');
@@ -166,5 +184,5 @@ export function createDev({
   }
   const idle = () => Promise.all([...entries.values()].map((e) => e.promise)).then(() => {});
 
-  return { ensure, touch, stop, list, status, sweep, cleanupOrphans, idle };
+  return { ensure, register, wake, touch, stop, list, status, sweep, cleanupOrphans, idle };
 }

runner/src/index.js

@@ -3,8 +3,8 @@ import { createApiServer } from './server.js';
 import { createDocker } from './docker.js';
 import { createJobs } from './jobs.js';
 import { createDev } from './dev.js';
+import { createProxy } from './proxy.js';
 
-// Proxy kommt in einem späteren Task.
 const config = loadConfig(process.env);
 const docker = createDocker();
 const realJobs = createJobs({ config, docker });
@@ -19,3 +19,6 @@ setInterval(() => dev.sweep().catch((e) => console.error('sweep', e)), 60e3).unr
 createApiServer({ config, jobs, dev }).listen(config.apiPort, () => {
   console.log(`Preview-Runner API auf Port ${config.apiPort}`);
 });
+createProxy({ config, dev }).listen(config.proxyPort, () => {
+  console.log(`Preview-Runner Proxy auf Port ${config.proxyPort}`);
+});

runner/src/proxy.js

@@ -0,0 +1,126 @@
+// Reverse-Proxy für Dev-Previews: Host <sub>.<baseHost> ⇒ laufender Dev-Container (R5: Docker-DNS im Netz preview-sandbox).
+// Keine Authentifizierung hier: die Hosts liegen hinter Traefiks preview-chain (ForwardAuth). Weitergeleitet wird
+// ausschließlich an das Ziel, das dev.wake/dev.ensure liefert (preview-dev-<sub>).
+import http from 'node:http';
+import { isValidSub } from './validate.js';
+
+const esc = (s) => String(s).replace(/[&<>"']/g, (c) => ({ '&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": '&#39;' }[c]));
+
+function page(title, text, { refresh = false, hint = '' } = {}) {
+  return `<!doctype html><html lang="de"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">${refresh ? '<meta http-equiv="refresh" content="3">' : ''}<title>${esc(title)}</title><style>body{margin:0;min-height:100vh;display:flex;align-items:center;justify-content:center;font-family:system-ui,sans-serif;background:#f5f6f8;color:#1c2330}main{max-width:26rem;padding:2rem;text-align:center}h1{font-size:1.25rem;margin:0 0 .5rem}p{margin:.25rem 0;color:#5b6575}</style></head><body><main><h1>${esc(title)}</h1><p>${esc(text)}</p>${hint ? `<p>${esc(hint)}</p>` : ''}</main></body></html>`;
+}
+
+const PAGES = {
+  starting: page('Preview startet …', 'Der Dev-Server wird gestartet. Die Seite lädt automatisch neu.', { refresh: true }),
+  failed: page('Preview konnte nicht starten', 'Der Dev-Server ist nicht hochgekommen.', { hint: 'Details stehen im Protokoll im WOMS.' }),
+  unknown: page('Preview nicht verfügbar', 'Diese Preview ist gerade nicht verfügbar.'),
+  busy: page('Preview nicht verfügbar', 'Der Server ist gerade ausgelastet. Bitte später erneut versuchen.'),
+};
+
+function html(res, status, body, extra = {}) {
+  res.writeHead(status, { 'Content-Type': 'text/html; charset=utf-8', 'Content-Length': Buffer.byteLength(body), 'Cache-Control': 'no-store', ...extra });
+  res.end(body);
+}
+
+export function createProxy({ config, dev, request = http.request }) {
+  // Sub aus dem Host-Header: exakt <sub>.<baseHost>, klein geschrieben, Port entfernt.
+  function subOf(hostHeader) {
+    const host = String(hostHeader || '').replace(/:\d+$/, '');
+    const suffix = `.${config.baseHost}`;
+    if (host !== host.toLowerCase() || !host.endsWith(suffix)) return null;
+    const sub = host.slice(0, -suffix.length);
+    return isValidSub(sub) ? sub : null;
+  }
+
+  function fwdHeaders(req) {
+    const h = { ...req.headers };
+    const prior = req.headers['x-forwarded-for'];
+    const ip = req.socket.remoteAddress || '';
+    h['x-forwarded-for'] = prior ? `${prior}, ${ip}` : ip;
+    h['x-forwarded-host'] = req.headers.host || '';
+    h['x-forwarded-proto'] = req.headers['x-forwarded-proto'] || 'https';
+    return h;
+  }
+
+  // Ergebnis: { target } wenn online, sonst { state }.
+  async function resolve(sub) {
+    let r;
+    try { r = await dev.wake(sub); } catch (e) { return { state: e?.status === 503 ? 'busy' : 'unknown' }; }
+    if (r.state === 'online' && r.target) return { target: new URL(r.target) };
+    return { state: r.state };
+  }
+
+  const server = http.createServer(async (req, res) => {
+    try {
+      const sub = subOf(req.headers.host);
+      if (!sub) return html(res, 404, page('Nicht gefunden', 'Diese Adresse gibt es nicht.'));
+      const r = await resolve(sub);
+      if (!r.target) {
+        const key = PAGES[r.state] ? r.state : 'unknown';
+        return html(res, 503, PAGES[key], key === 'starting' ? { 'Retry-After': '3' } : {});
+      }
+      dev.touch(sub);
+      const up = request({
+        hostname: r.target.hostname, port: r.target.port, method: req.method, path: req.url, headers: fwdHeaders(req),
+      }, (ures) => {
+        res.writeHead(ures.statusCode, ures.headers);
+        ures.pipe(res);
+        ures.on('error', () => res.destroy());
+      });
+      up.on('error', () => {
+        if (!res.headersSent) {
+          const body = 'Die Preview ist gerade nicht erreichbar (502).';
+          res.writeHead(502, { 'Content-Type': 'text/plain; charset=utf-8', 'Content-Length': Buffer.byteLength(body) });
+          res.end(body);
+        } else res.destroy();
+      });
+      res.on('close', () => { if (!res.writableFinished) up.destroy(); });
+      req.pipe(up);
+    } catch (e) {
+      console.error('Proxy-Fehler:', e);
+      if (!res.headersSent) html(res, 500, page('Fehler', 'Interner Fehler.'));
+      else res.destroy();
+    }
+  });
+
+  server.on('upgrade', async (req, socket, head) => {
+    socket.on('error', () => socket.destroy());
+    try {
+      const sub = subOf(req.headers.host);
+      if (!sub) return socket.end('HTTP/1.1 404 Not Found\r\nConnection: close\r\n\r\n');
+      const r = await resolve(sub);
+      if (!r.target) return socket.end('HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\n\r\n');
+      dev.touch(sub);
+      const up = request({
+        hostname: r.target.hostname, port: r.target.port, method: req.method, path: req.url, headers: fwdHeaders(req),
+      });
+      up.on('response', (ures) => { // Upstream lehnt das Upgrade ab: Antwort durchreichen und schließen
+        let raw = `HTTP/1.1 ${ures.statusCode} ${ures.statusMessage}\r\n`;
+        for (let i = 0; i < ures.rawHeaders.length; i += 2) raw += `${ures.rawHeaders[i]}: ${ures.rawHeaders[i + 1]}\r\n`;
+        socket.write(`${raw}\r\n`);
+        ures.pipe(socket);
+      });
+      up.on('upgrade', (ures, usock, uhead) => {
+        usock.on('error', () => { usock.destroy(); socket.destroy(); });
+        // allowHalfOpen-Sockets bleiben nach 'end' offen: bei Ende oder Schließen einer Seite beide zerstören
+        const both = () => { usock.destroy(); socket.destroy(); };
+        for (const ev of ['end', 'close']) { socket.on(ev, both); usock.on(ev, both); }
+        let raw = `HTTP/1.1 ${ures.statusCode} ${ures.statusMessage}\r\n`;
+        for (let i = 0; i < ures.rawHeaders.length; i += 2) raw += `${ures.rawHeaders[i]}: ${ures.rawHeaders[i + 1]}\r\n`;
+        socket.write(`${raw}\r\n`);
+        if (uhead?.length) socket.write(uhead);
+        if (head?.length) usock.write(head);
+        usock.pipe(socket);
+        socket.pipe(usock);
+      });
+      up.on('error', () => socket.destroy());
+      socket.on('close', () => up.destroy());
+      up.end();
+    } catch (e) {
+      console.error('Proxy-Upgrade-Fehler:', e);
+      socket.destroy();
+    }
+  });
+
+  return server;
+}

runner/src/server.js

@@ -67,6 +67,14 @@ export function createApiServer({ config, jobs, dev, mem = readMemAvailable }) {
         const run = jobs.get(parts[1]);
         return run ? send(res, 200, run) : send(res, 404, { error: 'Lauf nicht gefunden' });
       }
+      if (method === 'PUT' && parts[0] === 'dev' && parts.length === 2) {
+        const sub = parts[1];
+        if (!isValidSub(sub)) return send(res, 400, { error: 'Ungültige Subdomain' });
+        const v = validateJob(await readJson(req));
+        if (!v.ok) return send(res, 400, { error: v.error });
+        if (v.job.sub !== sub) return send(res, 400, { error: 'Subdomain stimmt nicht überein' });
+        return send(res, 200, { state: (await dev.register(sub, v.job)).state });
+      }
       if (method === 'POST' && parts[0] === 'dev' && parts.length === 3 && ['start', 'stop'].includes(parts[2])) {
         const sub = parts[1];
         if (!isValidSub(sub)) return send(res, 400, { error: 'Ungültige Subdomain' });

runner/test/dev.test.js

@@ -249,3 +249,45 @@ test('cleanupOrphans: ein fehlschlagendes rm stoppt die übrigen nicht', async (
   try { await s.dev.cleanupOrphans(); } finally { console.error = orig; }
   assert.deepEqual(s.removed, ['preview-dev-x', 'preview-dev-y']);
 });
+
+test('register: speichert Spezifikation ohne Start (sleeping); wake startet damit', async () => {
+  const s = setup();
+  assert.deepEqual(await s.dev.register('a', job('a')), { state: 'sleeping' });
+  assert.equal(s.runs.length, 0); assert.equal(s.prepared.length, 0);
+  assert.deepEqual(await s.dev.wake('a'), { state: 'starting' });
+  await s.dev.idle();
+  assert.equal(s.runs.length, 1);
+  assert.deepEqual(await s.dev.wake('a'), { state: 'online', target: 'http://preview-dev-a:5173' });
+});
+
+test('wake: unbekannter sub ⇒ unknown, nach stop ebenfalls', async () => {
+  const s = setup();
+  assert.deepEqual(await s.dev.wake('x'), { state: 'unknown' });
+  await s.dev.register('a', job('a'));
+  await s.dev.stop('a');
+  assert.deepEqual(await s.dev.wake('a'), { state: 'unknown' });
+});
+
+test('register: läuft mit gleichem sha ⇒ unverändert; mit anderem sha ⇒ Neustart; wake nutzt neue Spezifikation', async () => {
+  const s = setup();
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  assert.deepEqual(await s.dev.register('a', job('a')), { state: 'online' });
+  assert.equal(s.runs.length, 1);
+  assert.deepEqual(await s.dev.register('a', job('a', { sha: '1234567' })), { state: 'starting' });
+  await s.dev.idle();
+  assert.equal(s.runs.length, 2);
+  assert.equal(s.prepared[1].job.sha, '1234567');
+  assert.ok(s.removed.includes('preview-dev-a'));
+});
+
+test('register: schlafender sub mit neuem sha startet nicht, wake nutzt das neue sha', async () => {
+  const s = setup();
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  s.clock.t += 21 * MIN; await s.dev.sweep();
+  assert.equal(s.dev.status('a').state, 'sleeping');
+  assert.deepEqual(await s.dev.register('a', job('a', { sha: '7654321' })), { state: 'sleeping' });
+  assert.equal(s.runs.length, 1);
+  await s.dev.wake('a'); await s.dev.idle();
+  assert.equal(s.runs.length, 2);
+  assert.equal(s.prepared[1].job.sha, '7654321');
+});

runner/test/proxy.test.js

@@ -0,0 +1,143 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import http from 'node:http';
+import net from 'node:net';
+import crypto from 'node:crypto';
+import { createProxy } from '../src/proxy.js';
+
+const config = { baseHost: 'project.webklar.com' };
+
+// Test-Ziel: HTTP-Echo + WebSocket-Echo per rohem Upgrade
+function startTarget() {
+  const seen = [];
+  const srv = http.createServer((req, res) => {
+    const chunks = [];
+    req.on('data', (c) => chunks.push(c));
+    req.on('end', () => {
+      seen.push({ method: req.method, url: req.url, headers: req.headers });
+      res.writeHead(201, { 'X-Test': 'ja' });
+      res.end(`${req.method} ${req.url} ${Buffer.concat(chunks).toString()}`);
+    });
+  });
+  srv.on('upgrade', (req, socket) => {
+    const acc = crypto.createHash('sha1').update(`${req.headers['sec-websocket-key']}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64');
+    socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${acc}\r\n\r\n`);
+    socket.on('data', (d) => socket.write(d)); // Echo
+    socket.on('end', () => socket.end());
+    socket.on('error', () => {});
+  });
+  return new Promise((r) => srv.listen(0, '127.0.0.1', () => r({ srv, seen, port: srv.address().port })));
+}
+
+async function withProxy(devImpl, fn, opts = {}) {
+  const touched = [];
+  const dev = { touch: (s) => touched.push(s), ...devImpl };
+  const proxy = createProxy({ config, dev, ...opts });
+  await new Promise((r) => proxy.listen(0, '127.0.0.1', r));
+  try { await fn({ port: proxy.address().port, touched }); } finally { proxy.closeAllConnections?.(); await new Promise((r) => proxy.close(r)); }
+}
+
+const get = (port, host, { method = 'GET', path = '/', body } = {}) => new Promise((resolve, reject) => {
+  const req = http.request({ port, host: '127.0.0.1', method, path, headers: { Host: host } }, (res) => {
+    const c = []; res.on('data', (d) => c.push(d)); res.on('end', () => resolve({ res, text: Buffer.concat(c).toString() }));
+  });
+  req.on('error', reject);
+  req.end(body);
+});
+
+test('online: Durchreichen inkl. POST-Body, X-Forwarded-*, Host, touch', async () => {
+  const t = await startTarget();
+  try {
+    await withProxy({ wake: async (s) => ({ state: 'online', target: `http://127.0.0.1:${t.port}` }) }, async ({ port, touched }) => {
+      const { res, text } = await get(port, 'demo.project.webklar.com:443', { method: 'POST', path: '/a?b=1', body: 'hallo' });
+      assert.equal(res.statusCode, 201);
+      assert.equal(res.headers['x-test'], 'ja');
+      assert.equal(text, 'POST /a?b=1 hallo');
+      assert.deepEqual(touched, ['demo']);
+      const h = t.seen[0].headers;
+      assert.equal(h.host, 'demo.project.webklar.com:443');
+      assert.equal(h['x-forwarded-host'], 'demo.project.webklar.com:443');
+      assert.equal(h['x-forwarded-proto'], 'https');
+      assert.ok(h['x-forwarded-for']);
+    });
+  } finally { t.srv.closeAllConnections?.(); await new Promise((r) => t.srv.close(r)); }
+});
+
+test('WebSocket-Upgrade wird durchgereicht (Echo)', async () => {
+  const t = await startTarget();
+  try {
+    await withProxy({ wake: async () => ({ state: 'online', target: `http://127.0.0.1:${t.port}` }) }, async ({ port, touched }) => {
+      const sock = net.connect(port, '127.0.0.1');
+      sock.on('error', () => {});
+      let buf = '';
+      const got = new Promise((resolve) => sock.on('data', (d) => { buf += d.toString('latin1'); if (buf.includes('\r\n\r\n') && buf.endsWith('ping-pong')) resolve(); }));
+      sock.write('GET /ws HTTP/1.1\r\nHost: demo.project.webklar.com\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n');
+      await new Promise((r) => { const i = setInterval(() => { if (buf.includes('\r\n\r\n')) { clearInterval(i); r(); } }, 5); });
+      assert.match(buf, /^HTTP\/1\.1 101/);
+      try {
+        sock.write('ping-pong');
+        await got;
+        assert.ok(buf.endsWith('ping-pong'));
+        assert.deepEqual(touched, ['demo']);
+      } finally { sock.destroy(); }
+    });
+  } finally { t.srv.closeAllConnections?.(); await new Promise((r) => t.srv.close(r)); }
+});
+
+test('starting ⇒ 503 mit Text, Retry-After und Meta-Refresh', async () => {
+  await withProxy({ wake: async () => ({ state: 'starting' }) }, async ({ port, touched }) => {
+    const { res, text } = await get(port, 'demo.project.webklar.com');
+    assert.equal(res.statusCode, 503);
+    assert.equal(res.headers['retry-after'], '3');
+    assert.match(text, /Preview startet …/);
+    assert.match(text, /<meta http-equiv="refresh" content="3">/);
+    assert.deepEqual(touched, []);
+  });
+});
+
+test('failed ⇒ 503 mit Fehlerseite', async () => {
+  await withProxy({ wake: async () => ({ state: 'failed' }) }, async ({ port }) => {
+    const { res, text } = await get(port, 'demo.project.webklar.com');
+    assert.equal(res.statusCode, 503);
+    assert.match(text, /Preview konnte nicht starten/);
+    assert.match(text, /Protokoll im WOMS/);
+    assert.ok(!text.includes('http-equiv="refresh"'));
+  });
+});
+
+test('wake unknown ⇒ 503 „nicht verfügbar“', async () => {
+  await withProxy({ wake: async () => ({ state: 'unknown' }) }, async ({ port }) => {
+    const { res, text } = await get(port, 'demo.project.webklar.com');
+    assert.equal(res.statusCode, 503);
+    assert.match(text, /Diese Preview ist gerade nicht verfügbar\./);
+  });
+});
+
+test('Speicherschutz (wake wirft 503) ⇒ 503-Seite, kein Absturz', async () => {
+  await withProxy({ wake: async () => { throw Object.assign(new Error('Server ausgelastet'), { status: 503 }); } }, async ({ port }) => {
+    const { res, text } = await get(port, 'demo.project.webklar.com');
+    assert.equal(res.statusCode, 503);
+    assert.match(text, /ausgelastet/);
+  });
+});
+
+test('fremder Host, Großschreibung, ungültiger sub, mehrstufig ⇒ 404 ohne wake', async () => {
+  let woken = 0;
+  await withProxy({ wake: async () => { woken++; return { state: 'starting' }; } }, async ({ port }) => {
+    for (const h of ['evil.example.com', 'project.webklar.com', 'Demo.project.webklar.com', 'a_b.project.webklar.com', 'a.b.project.webklar.com', '.project.webklar.com', 'demo.project.webklar.com.evil.com', '']) {
+      const { res } = await get(port, h);
+      assert.equal(res.statusCode, 404, h);
+    }
+    assert.equal(woken, 0);
+  });
+});
+
+test('Upstream-Fehler ⇒ 502 deutscher Text', async () => {
+  const dead = net.createServer(); await new Promise((r) => dead.listen(0, '127.0.0.1', r));
+  const port0 = dead.address().port; await new Promise((r) => dead.close(r));
+  await withProxy({ wake: async () => ({ state: 'online', target: `http://127.0.0.1:${port0}` }) }, async ({ port }) => {
+    const { res, text } = await get(port, 'demo.project.webklar.com');
+    assert.equal(res.statusCode, 502);
+    assert.match(text, /nicht erreichbar/);
+  });
+});

runner/test/server.test.js

@@ -17,8 +17,10 @@ const jobs = {
   queueLength: () => 0,
 };
 const stopped = [];
+const registered = [];
 const dev = {
   ensure: async (sub) => ({ sub, state: 'starting' }),
+  register: async (sub, j) => { registered.push([sub, j]); return { state: 'sleeping', extra: 'x' }; },
   stop: async (sub) => { stopped.push(sub); },
   list: () => [],
 };
@@ -133,3 +135,19 @@ test('readMemAvailable liest MemAvailable in Bytes', () => {
   writeFileSync(p, 'MemTotal:       8000000 kB\nMemFree:         100000 kB\nMemAvailable:    2048000 kB\n');
   assert.equal(readMemAvailable(p), 2048000 * 1024);
 });
+
+test('PUT /dev/:sub: Auth, 400, 200 ⇒ register', async () => {
+  await withServer(config, async (base) => {
+    const put = (path, headers, body) => fetch(base + path, { method: 'PUT', headers, body: JSON.stringify(body) });
+    assert.equal((await put('/dev/demo', { 'Content-Type': 'application/json' }, good)).status, 401);
+    assert.equal((await put('/dev/demo', auth, { ...good, sub: 'andere' })).status, 400);
+    assert.equal((await put('/dev/demo', auth, { ...good, sha: 'zz' })).status, 400);
+    assert.equal((await put('/dev/DEMO', auth, good)).status, 400);
+    assert.equal(registered.length, 0);
+    const r = await put('/dev/demo', auth, good);
+    assert.equal(r.status, 200);
+    assert.deepEqual(await r.json(), { state: 'sleeping' });
+    assert.equal(registered[0][0], 'demo');
+    assert.equal(registered[0][1].repo, 'JUSN/demo');
+  });
+});