import { createServer } from "node:http";
import { readFile } from "node:fs/promises";
import { join } from "node:path";
import crypto from "node:crypto";
import { expect, test } from "@playwright/test";
import { PGlite } from "@electric-sql/pglite";
import type { SupabaseClient } from "@supabase/supabase-js";
import { streamText, tool } from "ai";
import { MockLanguageModelV4 } from "ai/test";
import { z } from "zod";
import { createPostgrestShim } from "../../lib/db/postgrest-shim";
import {
  createSlackWebhookPostHandler,
  buildSlackEventTaskPayload,
} from "../../app/api/webhooks/slack/route";
import { runSlackEventTask } from "../../trigger/slack-event";
import { resolveKnownSlackAttribution } from "../../trigger/slack-event-lib/attribution";
import {
  findSlackGuidanceRuns,
  submitSlackRunGuidance,
  loadRunGuidance,
  deliverRunGuidance,
} from "../../lib/slack/run-guidance-store";
import { createRunGuidanceSession } from "../../lib/slack/run-guidance-session";
import { deliverSlackRunUpdate } from "../../lib/slack/run-delivery";
import { stripSlackRunControlsForTerminalRun } from "../../lib/slack/run-controls-notify";
import { emptyRunResultEvidence } from "../../lib/slack/run-result-evidence";
import { buildRunRow } from "../unit/helpers/mogplex-api-runs-fixtures";
import type {
  PostSlackMessageInput,
  UpdateSlackMessageInput,
} from "../../lib/slack/client";
import type { SlackInstallationRow } from "../../lib/slack/installations";

// Service E2E: actual signed HTTP webhook, event routing, Postgres inbox,
// SDK step boundaries and Slack message writer. Only provider/queue transports
// and pre-existing account/installation records are fixtures. Native Slack
// rendering is separately verified in a real workspace, not imitated in HTML.
test("a Slack thread reply reaches the next agent step once and survives terminal delivery", async ({
  request,
}) => {
  const owner = "00000000-0000-4000-8000-000000000001";
  const runId = "00000000-0000-4000-8000-000000000002";
  const callId = "00000000-0000-4000-8000-000000000003";
  const run = buildRunRow({
    id: runId,
    user_id: owner,
    ai_call_id: callId,
    harness: "mogplex",
    status: "streaming",
    prompt: "Fix the mobile controls",
    metadata: {
      slack_guidance_enabled: true,
      slack_user_id: "U1",
      slack_thread_ts: "1.2",
      slackRunControls: { teamId: "T1", channelId: "D1", messageTs: "1.2" },
    },
  });
  const db = await PGlite.create();
  await db.exec(`create role anon; create role authenticated; create role service_role;
    create table profiles(id uuid primary key); create table ai_calls(id uuid primary key);
    create table external_agent_runs(id uuid primary key, user_id uuid, ai_call_id uuid, status text, harness text, metadata jsonb, slack_progress jsonb, slack_progress_revision bigint not null default 0);`);
  await db.exec(
    await readFile(
      join(
        process.cwd(),
        "supabase/migrations/20260905124500_slack_run_guidance.sql"
      ),
      "utf8"
    )
  );
  // Deploys apply every migration before the code that reads its columns.
  await db.exec(
    await readFile(
      join(
        process.cwd(),
        "neon/migrations/20261026021000_run_guidance_replies.sql"
      ),
      "utf8"
    )
  );
  await db.query("insert into profiles values($1)", [owner]);
  await db.query("insert into ai_calls values($1)", [callId]);
  await db.query(
    "insert into external_agent_runs(id,user_id,ai_call_id,status,harness,metadata) values($1,$2,$3,'streaming','mogplex',$4)",
    [runId, owner, callId, JSON.stringify(run.metadata)]
  );
  const client = createPostgrestShim({
    query: async (sql, values) => ({
      rows: (await db.query(sql, values)).rows as Record<string, unknown>[],
    }),
  }) as unknown as SupabaseClient;
  const messages: UpdateSlackMessageInput[] = [];
  const announcements: PostSlackMessageInput[] = [];
  const acknowledgements: string[] = [];
  let unexpectedAgentStarts = 0;
  const updates: Array<{ runId: string; userId: string }> = [];
  const installation: SlackInstallationRow = {
    id: "installation-1",
    team_id: "T1",
    team_name: "Fixture",
    installed_by_user_id: owner,
    bot_user_id: "UBOT",
    vault_bot_token_id: "fixture",
    scopes: ["chat:write"],
    authed_user_slack_id: "U1",
    created_at: new Date(0).toISOString(),
    updated_at: new Date(0).toISOString(),
  };
  const secret = crypto.randomBytes(32).toString("hex");
  const handler = createSlackWebhookPostHandler({
    getSigningSecret: () => secret,
    dispatch: async (event) => {
      if (event.kind !== "event") throw new Error("Expected a Slack event");
      const payload = buildSlackEventTaskPayload(event.body);
      if (!payload) throw new Error("Missing webhook identity");
      await runSlackEventTask(payload, {
        getInstallation: async (teamId) =>
          teamId === installation.team_id ? installation : null,
        getBotToken: async () => "fixture-token",
        resolveSlackAttribution: async (installed, slackUserId) =>
          resolveKnownSlackAttribution({
            installation: installed,
            slackUserId,
            existing: null,
          }) ?? { mode: "unmapped", mogplexUserId: null, slackEmail: null },
        findGuidanceRuns: (input) => findSlackGuidanceRuns(input, client),
        submitGuidance: (input) => submitSlackRunGuidance(input, client),
        queueRunDelivery: async (input) => {
          updates.push(input);
        },
        postMessage: async (_token, input) => {
          acknowledgements.push(input.text);
          return { channel: "D1", ts: "1.4" };
        },
        startRepoAgentRun: async () => {
          unexpectedAgentStarts++;
          throw new Error("A guidance reply must not launch a run");
        },
        runAgent: async () => {
          unexpectedAgentStarts++;
          throw new Error("A guidance reply must not launch a chat agent");
        },
      });
    },
  });
  const server = createServer(async (req, res) => {
    try {
      const chunks: Buffer[] = [];
      for await (const chunk of req) chunks.push(Buffer.from(chunk));
      const result = await handler(
        new Request("http://localhost/api/webhooks/slack", {
          method: "POST",
          headers: req.headers as Record<string, string>,
          body: Buffer.concat(chunks).toString(),
        })
      );
      res.writeHead(result.status, { "content-type": "application/json" });
      res.end(await result.text());
    } catch {
      res.writeHead(500);
      res.end();
    }
  });
  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
  const address = server.address();
  if (!address || typeof address === "string")
    throw new Error("Missing fixture address");
  const endpoint = `http://127.0.0.1:${address.port}/api/webhooks/slack`;
  const raw = JSON.stringify({
    type: "event_callback",
    team_id: "T1",
    event_id: "Ev-guide",
    event: {
      type: "message",
      channel_type: "im",
      channel: "D1",
      user: "U1",
      ts: "1.3",
      thread_ts: "1.2",
      text: "Keep the desktop header unchanged.",
    },
  });
  const timestamp = String(Math.floor(Date.now() / 1000));
  const headers = {
    "content-type": "application/json",
    "x-slack-request-timestamp": timestamp,
    "x-slack-signature":
      "v0=" +
      crypto
        .createHmac("sha256", secret)
        .update(`v0:${timestamp}:${raw}`)
        .digest("hex"),
  };
  const previousAppUrl = process.env.NEXT_PUBLIC_APP_URL;
  process.env.NEXT_PUBLIC_APP_URL = "http://localhost:3000";
  try {
    expect(
      (
        await request.post(endpoint, {
          data: raw,
          headers: { ...headers, "x-slack-signature": "v0=invalid" },
        })
      ).status()
    ).toBe(401);
    expect(await loadRunGuidance(run, client)).toEqual([]);
    const session = createRunGuidanceSession(run, {
      load: (row) => loadRunGuidance(row, client),
      deliver: (input) => deliverRunGuidance(input, client),
      queue: async (input) => {
        updates.push(input);
      },
    });
    let step = 0;
    const finalReport =
      "Desktop controls preserved. " + "Inspected the layout. ".repeat(100);
    const model = new MockLanguageModelV4({
      doStream: async () => ({
        stream: new ReadableStream({
          start(sink) {
            if (step++ < 2)
              sink.enqueue({
                type: "tool-call",
                toolCallId: `command-${step}`,
                toolName: "inspect",
                input: "{}",
              });
            else {
              sink.enqueue({ type: "text-start", id: "result" });
              sink.enqueue({
                type: "text-delta",
                id: "result",
                delta: finalReport,
              });
              sink.enqueue({ type: "text-end", id: "result" });
            }
            sink.enqueue({
              type: "finish",
              finishReason: {
                unified: step <= 2 ? "tool-calls" : "stop",
                raw: "stop",
              },
              usage: {
                inputTokens: {
                  total: 1,
                  noCache: 1,
                  cacheRead: 0,
                  cacheWrite: 0,
                },
                outputTokens: { total: 1, text: 1, reasoning: 0 },
              },
            });
            sink.close();
          },
        }),
      }),
    });
    const result = streamText({
      model,
      prompt: run.prompt,
      stopWhen: () => false,
      prepareStep: async ({ messages, stepNumber }) => ({
        messages: await session.prepare(messages, stepNumber),
      }),
      onStepEnd: () => session.stepFinished(),
      tools: {
        inspect: tool({
          inputSchema: z.object({}),
          execute: async () => {
            if (step === 1) {
              expect(
                (await request.post(endpoint, { data: raw, headers })).status()
              ).toBe(200);
              expect((await loadRunGuidance(run, client))[0].status).toBe(
                "received"
              );
            }
            return "Inspected the header";
          },
        }),
      },
    });
    await result.consumeStream();
    expect(await result.text).toBe(finalReport);
    expect(JSON.stringify(model.doStreamCalls[0].prompt)).not.toContain(
      "Keep the desktop header unchanged."
    );
    expect(JSON.stringify(model.doStreamCalls[1].prompt)).toContain(
      "Keep the desktop header unchanged."
    );
    expect(model.doStreamCalls).toHaveLength(3);
    expect(
      JSON.stringify(model.doStreamCalls[2].prompt).match(
        /Keep the desktop header unchanged\./g
      )
    ).toHaveLength(1);
    expect((await loadRunGuidance(run, client))[0]).toMatchObject({
      status: "delivered",
      delivered_step: 1,
    });
    await db.query(
      "update external_agent_runs set status='success' where id=$1",
      [runId]
    );
    const finished = {
      ...run,
      status: "success" as const,
      slack_progress_delivered_at: new Date().toISOString(),
    };
    await deliverSlackRunUpdate(
      { runId, userId: owner },
      {
        loadRun: async () => finished,
        wait: async () => {
          throw new Error("Terminal delivery must bypass progress pacing");
        },
        sendTerminal: (row, status, _deps, options) =>
          stripSlackRunControlsForTerminalRun(
            row,
            status,
            {
              getSlackBotToken: async () => "fixture-token",
              updateSlackMessage: async (_token, message) => {
                messages.push(message);
              },
              postSlackMessage: async (_token, message) => {
                announcements.push(message);
              },
              loadRunOutput: async () => result.text,
              loadGuidance: async () => loadRunGuidance(run, client),
              loadEvidence: async () => emptyRunResultEvidence(),
            },
            options
          ),
        markDelivered: async () => {},
      }
    );
    expect(messages[0].text).toContain("Supplied to agent step 2");
    expect(messages[0].text).toContain(
      "*Agent’s closing report (excerpt)*\nDesktop controls preserved."
    );
    expect(announcements).toHaveLength(1);
    expect(announcements[0].text).toContain("*✅ Run finished:*");
    expect(JSON.stringify(messages[0].blocks)).not.toContain(
      "mogplex-cancel-run"
    );
    expect(
      (await request.post(endpoint, { data: raw, headers })).status()
    ).toBe(200);
    expect(await loadRunGuidance(run, client)).toHaveLength(1);
    expect(acknowledgements.at(-1)).toContain("already supplied");
    expect(updates).toHaveLength(3);
    expect(unexpectedAgentStarts).toBe(0);
  } finally {
    if (previousAppUrl === undefined) delete process.env.NEXT_PUBLIC_APP_URL;
    else process.env.NEXT_PUBLIC_APP_URL = previousAppUrl;
    await new Promise<void>((resolve, reject) =>
      server.close((error) => (error ? reject(error) : resolve()))
    );
    await db.close();
  }
});