import { CommandId } from "@t3tools/contracts";import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";import * as Cause from "effect/Cause";import * as Context from "effect/Context";import * as Crypto from "effect/Crypto";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Schedule from "effect/Schedule";import type * as Scope from "effect/Scope";import * as Stream from "effect/Stream"; import * as GitManager from "../git/GitManager.ts";import * as PullRequestService from "../pullRequest/PullRequestService.ts";import * as ServerSettings from "../serverSettings.ts";import { forkParked } from "../serverActivation.ts";import * as OrchestrationEngine from "./Services/OrchestrationEngine.ts";import * as ProjectionSnapshotQuery from "./Services/ProjectionSnapshotQuery.ts";import {  isAutoSettlementCandidate,  resolveAutoSettlementAt,  type SettlementPullRequest,} from "./ThreadSettlementPolicy.ts"; export class ThreadSettlementReactor extends Context.Service<  ThreadSettlementReactor,  {    readonly start: () => Effect.Effect<void, never, Scope.Scope>;    readonly drain: Effect.Effect<void>;  }>()("t3/orchestration/ThreadSettlementReactor") {} export const make = Effect.gen(function* () {  const engine = yield* OrchestrationEngine.OrchestrationEngineService;  const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;  const settingsService = yield* ServerSettings.ServerSettingsService;  const git = yield* GitManager.GitManager;  const pullRequests = yield* PullRequestService.PullRequestService;  const crypto = yield* Crypto.Crypto;  const fileSystem = yield* FileSystem.FileSystem;   const sweep = Effect.fn("ThreadSettlementReactor.sweep")(function* (    mergedPullRequest: PullRequestService.PullRequestMergeEvent | null,  ) {    const snapshot = yield* snapshots.getShellSnapshot();    const now = DateTime.formatIso(yield* DateTime.now);    const projects = new Map(snapshot.projects.map((project) => [project.id, project]));    // A merge event re-sweeps every candidate, not just the threads linked to    // the merged pull request: most threads carry no link and settle from    // their branch lookup, which would otherwise wait for the next minute's    // sweep on a possibly stale cached answer.    const candidates = snapshot.threads.filter((thread) => isAutoSettlementCandidate(thread, now));    // Use the same cwd as the sidebar so both paths share GitManager's PR cache.    const lookupCwdByThreadId = new Map<string, string>();    yield* Effect.forEach(      candidates,      (thread) =>        Effect.gen(function* () {          const project = projects.get(thread.projectId);          if (project === undefined || thread.linkedPullRequest != null) return;          const worktreeExists =            thread.worktreePath !== null &&            (yield* fileSystem.exists(thread.worktreePath).pipe(Effect.orElseSucceed(() => false)));          lookupCwdByThreadId.set(            thread.id,            worktreeExists && thread.worktreePath !== null              ? thread.worktreePath              : project.workspaceRoot,          );        }),      { concurrency: 8, discard: true },    );    if (mergedPullRequest !== null) {      // The merge just confirmed a terminal state the lookup caches can still      // call open (branch answers live two minutes, the sweep runs every      // minute). Drop the swept checkouts' cached answers so the merge settles      // its branch threads now instead of on a later sweep. Threads linked to      // the merged pull request settle from the event itself below and need no      // lookup, so they are absent from this map by construction.      const cwds = [...new Set(lookupCwdByThreadId.values())];      yield* Effect.forEach(cwds, (cwd) => git.invalidateStatus(cwd), {        concurrency: 8,        discard: true,      });    }    const lookupKey = (thread: (typeof candidates)[number]) => {      if (thread.linkedPullRequest != null) {        return JSON.stringify([          "linked",          thread.linkedPullRequest.projectId,          thread.linkedPullRequest.repository,          thread.linkedPullRequest.number,        ]);      }      if (thread.branch === null) return JSON.stringify(["none", thread.id]);      const cwd = lookupCwdByThreadId.get(thread.id);      return JSON.stringify(        cwd === undefined ? ["missing-project", thread.id] : ["branch", cwd, thread.branch],      );    };    const groups = Map.groupBy(candidates, lookupKey);     const pullRequestFor = Effect.fn("ThreadSettlementReactor.pullRequestFor")(function* (      thread: (typeof candidates)[number],    ) {      if (thread.linkedPullRequest != null) {        // The event carries the merged state, so only the threads linked to        // that exact pull request settle from it. Every other linked thread        // falls through to a fresh summary lookup below: the merge sweep        // covers all candidates, and an unrelated merge must never settle        // them.        if (          mergedPullRequest !== null &&          thread.linkedPullRequest.projectId === mergedPullRequest.projectId &&          thread.linkedPullRequest.repository.toLowerCase() ===            mergedPullRequest.repository.toLowerCase() &&          thread.linkedPullRequest.number === mergedPullRequest.number        ) {          return {            state: "merged",            updatedAt: mergedPullRequest.mergedAt,          } satisfies SettlementPullRequest;        }        if (!projects.has(thread.linkedPullRequest.projectId)) {          return yield* Effect.die(new Error("linked pull request project not found"));        }        const summary = yield* pullRequests.summary(          {            projectId: thread.linkedPullRequest.projectId,            repository: thread.linkedPullRequest.repository,            number: thread.linkedPullRequest.number,          },          { recoverTransientFailure: false },        );        return {          state: summary.state,          updatedAt: summary.updatedAt,        } satisfies SettlementPullRequest;      }      if (thread.branch === null) return null;      const cwd = lookupCwdByThreadId.get(thread.id);      if (cwd === undefined) {        return yield* Effect.die(new Error("thread project not found"));      }      return yield* git.branchPullRequest({ cwd, branch: thread.branch });    });     yield* Effect.forEach(      groups.values(),      (group) =>        Effect.gen(function* () {          const pullRequest = yield* pullRequestFor(group[0]!);          yield* Effect.forEach(            group,            (thread) =>              Effect.gen(function* () {                const settings = yield* settingsService.getSettings;                const decisionNow = DateTime.formatIso(yield* DateTime.now);                const settledAt = resolveAutoSettlementAt({                  thread,                  pullRequest,                  now: decisionNow,                  autoSettleAfterDays: settings.sidebarAutoSettleAfterDays,                  autoSettleOnMerge: settings.sidebarAutoSettleOnMerge,                });                if (settledAt === null) {                  return;                }                const uuid = yield* crypto.randomUUIDv4;                yield* engine.dispatch({                  type: "thread.auto-settle",                  commandId: CommandId.make(`server:auto-settle:${thread.id}:${uuid}`),                  threadId: thread.id,                  snapshotSequence: snapshot.snapshotSequence,                  settledAt,                });              }).pipe(                Effect.catchCause((cause) =>                  Cause.hasInterruptsOnly(cause)                    ? Effect.failCause(cause)                    : Effect.logWarning("automatic thread settlement skipped", {                        threadId: thread.id,                        cause: Cause.pretty(cause),                      }),                ),              ),            { discard: true },          );        }).pipe(          Effect.catchCause((cause) =>            Cause.hasInterruptsOnly(cause)              ? Effect.failCause(cause)              : Effect.logWarning("automatic thread settlement skipped", {                  threadIds: group.map((thread) => thread.id),                  cause: Cause.pretty(cause),                }),          ),        ),      { concurrency: 8, discard: true },    );  });   const runSweep = (mergedPullRequest: PullRequestService.PullRequestMergeEvent | null) =>    sweep(mergedPullRequest).pipe(      Effect.catchCause((cause) =>        Cause.hasInterruptsOnly(cause)          ? Effect.failCause(cause)          : Effect.logWarning("automatic thread settlement sweep failed", {              cause: Cause.pretty(cause),            }),      ),    );  const worker = yield* makeDrainableWorker(() => runSweep(null));   const start: ThreadSettlementReactor["Service"]["start"] = Effect.fn(    "ThreadSettlementReactor.start",  )(function* () {    const settingsChanges = yield* settingsService.subscribeChanges;    const mergedPullRequests = yield* pullRequests.subscribeMerges;    const initialSettings = yield* settingsService.getSettings.pipe(Effect.orDie);    let lastAfterDays = initialSettings.sidebarAutoSettleAfterDays;    let lastOnMerge = initialSettings.sidebarAutoSettleOnMerge;    yield* forkParked(      Effect.gen(function* () {        yield* worker.enqueue(undefined);        yield* worker.drain;      }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid),    );    yield* forkParked(      Stream.runForEach(settingsChanges, (settings) => {        if (          settings.sidebarAutoSettleAfterDays === lastAfterDays &&          settings.sidebarAutoSettleOnMerge === lastOnMerge        ) {          return Effect.void;        }        lastAfterDays = settings.sidebarAutoSettleAfterDays;        lastOnMerge = settings.sidebarAutoSettleOnMerge;        return worker.enqueue(undefined);      }),    );    yield* forkParked(Stream.runForEach(mergedPullRequests, runSweep));  });   return { start, drain: worker.drain } satisfies ThreadSettlementReactor["Service"];}); export const layer = Layer.effect(ThreadSettlementReactor, make);