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 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; readonly drain: Effect.Effect; } >()("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 sweep = Effect.fn("ThreadSettlementReactor.sweep")(function* () { const snapshot = yield* snapshots.getShellSnapshot(); const now = DateTime.formatIso(yield* DateTime.now); const projects = new Map(snapshot.projects.map((project) => [project.id, project])); const candidates = snapshot.threads.filter((thread) => isAutoSettlementCandidate(thread, now)); 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 project = projects.get(thread.projectId); return JSON.stringify( project === undefined ? ["missing-project", thread.id] : ["branch", project.workspaceRoot, thread.branch], ); }; const groups = Map.groupBy(candidates, lookupKey); const pullRequestFor = Effect.fn("ThreadSettlementReactor.pullRequestFor")(function* ( thread: (typeof candidates)[number], ) { if (thread.linkedPullRequest != null) { 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 project = projects.get(thread.projectId); if (project === undefined) { return yield* Effect.die(new Error("thread project not found")); } return yield* git.branchPullRequest({ cwd: project.workspaceRoot, 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 worker = yield* makeDrainableWorker(() => sweep().pipe( Effect.catchCause((cause) => Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) : Effect.logWarning("automatic thread settlement sweep failed", { cause: Cause.pretty(cause), }), ), ), ); const start: ThreadSettlementReactor["Service"]["start"] = Effect.fn( "ThreadSettlementReactor.start", )(function* () { const settingsChanges = yield* settingsService.subscribeChanges; 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); }), ); }); return { start, drain: worker.drain } satisfies ThreadSettlementReactor["Service"]; }); export const layer = Layer.effect(ThreadSettlementReactor, make);