pikasTech-HWLAB/internal/harnessrl/registry.ts

import pg from "pg";
import { readFile } from "node:fs/promises";

import { buildPostgresPoolConfig, postgresSslModeFromConnectionString } from "../db/runtime-store-core.ts";
import type { ActivityResult, CaseRunEvent, CaseRunRecord, CaseRunStatus, HarnessRLRegistry, ListCaseRunsInput, StartCaseRunInput } from "./contracts.ts";
import { terminalStatus } from "./contracts.ts";

const { Pool } = pg;

const migrationUrl = new URL("./migrations/0001_harnessrl_registry.sql", import.meta.url);

export class PostgresHarnessRLRegistry implements HarnessRLRegistry {
  private readonly pool: pg.Pool;
  private ready?: Promise<void>;

  constructor(databaseUrl: string) {
    if (!databaseUrl) throw codedError("harnessrl_database_url_required", "HARNESSRL_DATABASE_URL is required");
    this.pool = new Pool(buildPostgresPoolConfig({
      dbUrl: databaseUrl,
      sslMode: postgresSslModeFromConnectionString(databaseUrl) ?? "disable",
      poolMax: 8
    }));
  }

  ensureSchema() {
    this.ready ??= this.migrate();
    return this.ready;
  }

  private async migrate() {
    const client = await this.pool.connect();
    try {
      await client.query("BEGIN");
      await client.query(await readFile(migrationUrl, "utf8"));
      await client.query("COMMIT");
    } catch (error) {
      await client.query("ROLLBACK");
      throw error;
    } finally {
      client.release();
    }
  }

  async createRun(input: StartCaseRunInput) {
    await this.ensureSchema();
    const result = await this.pool.query(
      `INSERT INTO harnessrl_runs (run_id,case_id,workflow_id,status,stage,case_repo,run_dir,runtime_api_url)
       VALUES ($1,$2,$3,'queued','queued',$4,$5,$6)
       ON CONFLICT (run_id) DO UPDATE SET run_id=EXCLUDED.run_id
       RETURNING *`,
      [input.runId, input.caseId, input.workflowId, input.caseRepo, input.runDir, input.runtimeApiUrl]
    );
    return runRow(result.rows[0]);
  }

  async getRun(runId: string) {
    await this.ensureSchema();
    const result = await this.pool.query("SELECT * FROM harnessrl_runs WHERE run_id=$1", [runId]);
    return result.rows[0] ? runRow(result.rows[0]) : null;
  }

  async listRuns(input: ListCaseRunsInput) {
    await this.ensureSchema();
    const values: unknown[] = [];
    const where: string[] = [];
    if (input.status) {
      values.push(input.status);
      where.push(`status=$${values.length}`);
    }
    if (input.before) {
      values.push(input.before.createdAt, input.before.runId);
      where.push(`(created_at,run_id)<($${values.length - 1}::timestamptz,$${values.length})`);
    }
    values.push(input.limit);
    const result = await this.pool.query(
      `SELECT * FROM harnessrl_runs${where.length ? ` WHERE ${where.join(" AND ")}` : ""}
       ORDER BY created_at DESC,run_id DESC LIMIT $${values.length}`,
      values
    );
    return result.rows.map(runRow);
  }

  async updateRun(runId: string, patch: Partial<Pick<CaseRunRecord, "workflowRunId" | "status" | "stage" | "result" | "error" | "blocker">>) {
    await this.ensureSchema();
    const current = await this.getRun(runId);
    if (!current) throw codedError("caserun_run_not_found", "CaseRun run was not found");
    const result = await this.pool.query(
      `UPDATE harnessrl_runs SET workflow_run_id=$2,status=$3,stage=$4,result=$5,error=$6,blocker=$7,updated_at=now()
       WHERE run_id=$1 RETURNING *`,
      [runId, patch.workflowRunId === undefined ? current.workflowRunId : patch.workflowRunId,
        patch.status ?? current.status, patch.stage ?? current.stage,
        patch.result === undefined ? current.result : patch.result,
        patch.error === undefined ? current.error : patch.error,
        patch.blocker === undefined ? current.blocker : patch.blocker]
    );
    return runRow(result.rows[0]);
  }

  async appendEvent(runId: string, status: CaseRunStatus, stage: string, payload: Record<string, unknown> = {}) {
    await this.ensureSchema();
    const result = await this.pool.query(
      "INSERT INTO harnessrl_run_events (run_id,status,stage,payload) VALUES ($1,$2,$3,$4) RETURNING *",
      [runId, status, stage, payload]
    );
    return eventRow(result.rows[0]);
  }

  async listEvents(runId: string) {
    await this.ensureSchema();
    const result = await this.pool.query("SELECT * FROM harnessrl_run_events WHERE run_id=$1 ORDER BY id", [runId]);
    return result.rows.map(eventRow);
  }

  async getActivityResult(runId: string, activityName: string, identity: string) {
    await this.ensureSchema();
    const result = await this.pool.query(
      "SELECT * FROM harnessrl_activity_results WHERE run_id=$1 AND activity_name=$2 AND identity=$3",
      [runId, activityName, identity]
    );
    return result.rows[0] ? activityRow(result.rows[0]) : null;
  }

  async saveActivityResult(result: ActivityResult) {
    await this.ensureSchema();
    const saved = await this.pool.query(
      `INSERT INTO harnessrl_activity_results (run_id,activity_name,identity,output) VALUES ($1,$2,$3,$4)
       ON CONFLICT (run_id,activity_name,identity) DO UPDATE SET output=EXCLUDED.output RETURNING *`,
      [result.runId, result.activityName, result.identity, result.output]
    );
    return activityRow(saved.rows[0]);
  }

  async health() {
    await this.ensureSchema();
    await this.pool.query("SELECT 1");
    return { ok: true as const };
  }

  async close() { await this.pool.end(); }
}

function runRow(row: any): CaseRunRecord {
  const status = row.status as CaseRunStatus;
  return {
    runId: row.run_id, caseId: row.case_id, workflowId: row.workflow_id, workflowRunId: row.workflow_run_id,
    status, stage: row.stage, terminal: terminalStatus(status), caseRepo: row.case_repo, runDir: row.run_dir,
    runtimeApiUrl: row.runtime_api_url, result: row.result, error: row.error, blocker: row.blocker,
    createdAt: new Date(row.created_at).toISOString(), updatedAt: new Date(row.updated_at).toISOString()
  };
}

function eventRow(row: any): CaseRunEvent {
  return { id: Number(row.id), runId: row.run_id, status: row.status, stage: row.stage, payload: row.payload ?? {}, createdAt: new Date(row.created_at).toISOString() };
}

function activityRow(row: any): ActivityResult {
  return { runId: row.run_id, activityName: row.activity_name, identity: row.identity, output: row.output ?? {} };
}

function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }