// Shared: DB URLs, OpenWorkflow client, Kysely business DB (schema "biz"), workflow definitions.
import { OpenWorkflow } from "openworkflow";
import { BackendPostgres } from "openworkflow/postgres";
import { Kysely, PostgresDialect, sql } from "kysely";
import pg from "pg";

export const PG_URL = process.env.PG_URL ?? "postgresql://postgres:spike@127.0.0.1:55434/postgres";
export const WF_VERSION = process.env.WF_VERSION ?? "v1";
export const TAG = process.env.TAG ?? `pid${process.pid}`;

export const db = new Kysely<any>({
  dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: PG_URL, max: 4 }) }),
});

export async function initBiz() {
  await sql`CREATE SCHEMA IF NOT EXISTS biz`.execute(db);
  await sql`CREATE TABLE IF NOT EXISTS biz.side_effect_ledger (
    id bigserial PRIMARY KEY, run_id text, step text, worker text, pid int, note text,
    at timestamptz DEFAULT now())`.execute(db);
}

export async function ledger(runId: string, step: string, note = "") {
  await db.insertInto("biz.side_effect_ledger")
    .values({ run_id: runId, step, worker: TAG, pid: process.pid, note }).execute();
  console.log(`[${TAG}] run=${runId.slice(0, 8)} step=${step} ${note}`);
}

const wait = (ms: number) => new Promise((r) => setTimeout(r, ms));

export async function makeClient() {
  // connect() runs migrations by default (runMigrations: true), schema default "openworkflow".
  const backend = await BackendPostgres.connect(PG_URL, { schema: "openworkflow" });
  const ow = new OpenWorkflow({ backend });

  // Main vertical slice: submit download -> wait signal -> copy -> register -> retention sleep -> cleanup
  const acquire = ow.defineWorkflow(
    { name: "acquire", version: WF_VERSION },
    async ({ input, step, run, version }) => {
      const { downloadId, copyMs = 0, retention = "3s" } = input as any;
      const sub = await step.run({ name: "submit-download" }, async () => {
        await ledger(run.id, "submit-download", `version=${version}`);
        return { torrentHash: `hash-${downloadId}` };
      });
      const done = await step.waitForSignal<{ ok: boolean }>({
        name: "wait-download", signal: `download-done:${downloadId}`, timeout: "10m",
      });
      if (!done) throw new Error("download wait timed out");
      const copied = await step.run({ name: "copy-file" }, async () => {
        await ledger(run.id, "copy-file", "start");
        await wait(copyMs);
        await ledger(run.id, "copy-file", "end");
        return { dest: `/lib/${downloadId}.mkv` };
      });
      if (process.env.EXTRA_STEP === "1") {
        // EXTRA_STEP=1 simulates a code change between copy and register (tests code upgrade semantics)
        await step.run({ name: "v2-extra-verify" }, async () => { await ledger(run.id, "v2-extra-verify"); return 1; });
      }
      await step.run({ name: "register" }, async () => { await ledger(run.id, "register", `version=${version}`); return 1; });
      await step.sleep("retention", retention);
      await step.run({ name: "cleanup" }, async () => { await ledger(run.id, "cleanup"); return 1; });
      return { sub, copied, signal: done.data };
    },
  );

  // Retry test: step fails `failTimes` times then succeeds; step maximumAttempts = maxAttempts
  const flaky = ow.defineWorkflow(
    { name: "flaky", version: WF_VERSION },
    async ({ input, step, run }) => {
      const { failTimes, maxAttempts } = input as any;
      await step.run({ name: "before" }, async () => { await ledger(run.id, "before"); return 1; });
      return await step.run(
        { name: "flaky-step", retryPolicy: { maximumAttempts: maxAttempts, initialInterval: "200ms", maximumInterval: "1s" } },
        async () => {
          const n = await db.selectFrom("biz.side_effect_ledger").select(db.fn.countAll().as("c"))
            .where("run_id", "=", run.id).where("step", "=", "flaky-step").executeTakeFirst();
          const attempt = Number(n?.c ?? 0) + 1;
          await ledger(run.id, "flaky-step", `attempt=${attempt}`);
          if (attempt <= failTimes) throw new Error(`boom attempt ${attempt}`);
          return { attempt };
        },
      );
    },
  );

  // Cancel test: long step, then a follow-up step
  const slow = ow.defineWorkflow(
    { name: "slow", version: WF_VERSION },
    async ({ step, run }) => {
      await step.run({ name: "long" }, async () => {
        await ledger(run.id, "long", "start");
        await wait(8000);
        await ledger(run.id, "long", "end (still ran after cancel?)");
        return 1;
      });
      await step.run({ name: "after-long" }, async () => { await ledger(run.id, "after-long"); return 1; });
      return "done";
    },
  );

  // Concurrency test: many short runs
  const burst = ow.defineWorkflow(
    { name: "burst", version: WF_VERSION },
    async ({ step, run }) => {
      await step.run({ name: "work" }, async () => { await wait(300); await ledger(run.id, "work"); return 1; });
      return 1;
    },
  );

  return { ow, backend, acquire, flaky, slow, burst };
}