並列エージェントに実行環境をリースする — ポート帯・DBスキーマ・devサーバを1エージェント1環境で分離する設計

// lease.mjs — エージェント1体分の実行環境をリースする
import { writeFileSync } from 'node:fs';
import pg from 'pg';
 
// 1エージェントが使うポート数。dev(3000系)/api(4000系)/db-proxy(5000系)を
// まとめて1帯として確保する。10あれば周辺ツールの分も足りる。
const PORTS_PER_AGENT = 10;
const BASE = { dev: 3000, api: 4000, dbProxy: 5000 };
 
// エージェント番号(0,1,2,...)から決定的にポート帯を導く。
// 番号さえ固定すれば、再起動しても毎回同じポートに戻る点が重要。
function derivePorts(agentIndex) {
  const offset = agentIndex * PORTS_PER_AGENT;
  return {
    dev: BASE.dev + offset,       // 例: agent0=3000, agent1=3010
    api: BASE.api + offset,
    dbProxy: BASE.dbProxy + offset,
  };
}
 
// 専用スキーマを「移行済みテンプレート」から複製する。
// 毎回 migrate を走らせない。ここが並列化の最大のボトルネックだった。
async function provisionSchema(client, schema, template = 'tpl_migrated') {
  // 識別子は連結でSQLに埋め込むため、英数字とアンダースコアのみに制限する
  if (!/^[a-z0-9_]+$/.test(schema)) {
    throw new Error(`unsafe schema name: ${schema}`);
  }
  // 既存が残っていれば掃除してから、テンプレートの構造を複製する
  await client.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`);
  await client.query(`SELECT clone_schema('${template}', '${schema}')`);
  return schema;
}
 
export async function acquireLease(agentIndex, agentId) {
  const ports = derivePorts(agentIndex);
  const schema = `agent_${agentIndex}`;
 
  const client = new pg.Client({ connectionString: process.env.ADMIN_DATABASE_URL });
  await client.connect();
  try {
    await provisionSchema(client, schema);
  } finally {
    await client.end();
  }
 
  // search_path でスキーマを差し込む。アプリ側はテーブル名を変えなくてよい。
  const dbUrl =
    `${process.env.APP_DATABASE_URL}?options=-c%20search_path%3D${schema}`;
 
  const env = [
    `PORT=${ports.dev}`,
    `API_PORT=${ports.api}`,
    `DB_PROXY_PORT=${ports.dbProxy}`,
    `DATABASE_URL=${dbUrl}`,
    `AGENT_ID=${agentId}`,
    `LEASE_INDEX=${agentIndex}`,
  ].join('\n');
 
  writeFileSync(`.env.agent.${agentIndex}`, env + '\n');
  return { agentIndex, agentId, schema, ports, envFile: `.env.agent.${agentIndex}` };
}

-- 移行済みテンプレートの構造を、別スキーマに複製する関数。
-- テーブル定義・制約・シーケンスを引き継ぎ、データは空で始める。
CREATE OR REPLACE FUNCTION clone_schema(src text, dst text)
RETURNS void AS $$
DECLARE
  obj record;
BEGIN
  EXECUTE format('CREATE SCHEMA %I', dst);
 
  -- テーブル構造（制約・デフォルト込み）を複製。INCLUDING ALL が肝。
  FOR obj IN
    SELECT tablename FROM pg_tables WHERE schemaname = src
  LOOP
    EXECUTE format(
      'CREATE TABLE %I.%I (LIKE %I.%I INCLUDING ALL)',
      dst, obj.tablename, src, obj.tablename
    );
  END LOOP;
END;
$$ LANGUAGE plpgsql;

# エージェント2の帯(3020)に残ったプロセスを確実に掃除する。
# 決定的ポートだからこそ、掃除の対象が毎回同じで迷わない。
lsof -ti:3020 | xargs -r kill -9

// reap.mjs — 期限切れのリースを回収する（定期実行 or 起動時に走らせる）
import pg from 'pg';
import { execSync } from 'node:child_process';
 
const PORTS_PER_AGENT = 10;
const DEV_BASE = 3000;
 
export async function reapLease(agentIndex) {
  // 1) スキーマを落とす
  const client = new pg.Client({ connectionString: process.env.ADMIN_DATABASE_URL });
  await client.connect();
  try {
    await client.query(`DROP SCHEMA IF EXISTS agent_${agentIndex} CASCADE`);
  } finally {
    await client.end();
  }
 
  // 2) 帯に残ったプロセスを掃除する（決定的ポートなので対象が確定する）
  const devPort = DEV_BASE + agentIndex * PORTS_PER_AGENT;
  try {
    execSync(`lsof -ti:${devPort} | xargs -r kill -9`, { stdio: 'ignore' });
  } catch {
    // 掃除対象がなければ何もしない。ここで失敗しても回収は続行する
  }
}