ci-agent-runner/app/src/application/controller/handlers/workers.ts

import { rm } from "node:fs/promises";
import type { AgentStore } from "../../../adapters/database/store.js";
import type { ActorPolicy } from "../../../core/config.js";
import {
    parseCommentPayload,
    parseLabelPayload,
    type WebhookRepository,
} from "../../../core/webhook.js";
import { publishJob } from "../../publication/service.js";
import {
    claimJob,
    type PublicationContext,
    safeFailure,
    upsertJobStatus,
} from "../../publication/status.js";
import {
    IgnoreDelivery,
    reconcileCommand,
    reconcileLabels,
} from "./reconcile.js";

export async function pumpDeliveries(
    store: AgentStore,
    context: PublicationContext,
    repositoryId: number,
    repositoryFullName: string,
    botId: number,
    policy: ActorPolicy,
): Promise<void> {
    for (let count = 0; count < 20; count += 1) {
        const delivery = store.leaseDelivery();
        if (!delivery) break;
        try {
            if (delivery.eventType === "issue_label") {
                const payload = parseLabelPayload(delivery.payload);
                verifyIdentity(
                    payload.repository,
                    repositoryId,
                    repositoryFullName,
                );
                await reconcileLabels(
                    store,
                    context,
                    repositoryId,
                    payload.issue.number,
                    payload.sender,
                    botId,
                    policy,
                );
            } else if (delivery.eventType === "issue_comment") {
                const payload = parseCommentPayload(delivery.payload);
                verifyIdentity(
                    payload.repository,
                    repositoryId,
                    repositoryFullName,
                );
                await reconcileCommand(
                    store,
                    context,
                    repositoryId,
                    payload,
                    botId,
                    policy,
                );
            } else
                throw new IgnoreDelivery(
                    `Unsupported event type ${delivery.eventType}`,
                );
            store.completeDelivery(delivery.id);
        } catch (error) {
            if (error instanceof IgnoreDelivery) {
                store.completeDelivery(delivery.id);
                console.log(
                    log("info", "Ignored delivery", {
                        delivery: delivery.id,
                        reason: error.message,
                    }),
                );
            } else {
                store.retryDelivery(
                    delivery.id,
                    safeFailure(error),
                    delivery.attempts + 1,
                );
                console.error(
                    log("error", "Delivery processing failed", {
                        delivery: delivery.id,
                        error: safeFailure(error),
                    }),
                );
            }
        }
    }
}

export async function pumpOutbox(
    store: AgentStore,
    context: PublicationContext,
): Promise<void> {
    for (let count = 0; count < 10; count += 1) {
        const item = store.leaseOutbox();
        if (!item) break;
        const job = store.getJob(item.jobId);
        if (!job) {
            store.retryOutbox(item, "Outbox job no longer exists");
            continue;
        }
        try {
            if (item.kind === "claim") {
                if (job.cancelRequested)
                    await upsertJobStatus(
                        context.client,
                        context.botLogin,
                        job,
                        `Agent ${job.mode} cancelled`,
                        "The request was cancelled before execution.",
                    );
                else await claimJob(context, job);
                if (job.triggerLabel)
                    store.releaseLabelClaim(
                        job.repositoryId,
                        job.issueNumber,
                        job.triggerLabel,
                    );
                store.completeClaim(item);
            } else {
                const outcome = await publishJob(context, job);
                if (outcome.planCommentId !== undefined)
                    store.recordPlan(job, outcome.planCommentId);
                if (job.result?.implementation)
                    store.recordImplementation(
                        job,
                        outcome.commitSha || null,
                        outcome.pullRequestNumber || null,
                    );
                if (job.workspace)
                    await rm(job.workspace, {
                        recursive: true,
                        force: true,
                    }).catch((error) => {
                        console.error(
                            log("error", "Workspace cleanup failed", {
                                jobId: job.id,
                                error: safeFailure(error),
                            }),
                        );
                    });
                store.completePublication(item, outcome.terminal);
            }
        } catch (error) {
            store.retryOutbox(item, safeFailure(error));
            console.error(
                log("error", "Outbox operation failed", {
                    jobId: job.id,
                    kind: item.kind,
                    error: safeFailure(error),
                }),
            );
        }
    }
}

export function log(
    level: string,
    message: string,
    fields: Record<string, unknown>,
): string {
    return JSON.stringify({
        level,
        message,
        ...fields,
        time: new Date().toISOString(),
    });
}

function verifyIdentity(
    repository: WebhookRepository,
    id: number,
    fullName: string,
): void {
    if (
        repository.id !== id ||
        repository.full_name.toLowerCase() !== fullName.toLowerCase()
    ) {
        throw new IgnoreDelivery("Repository identity mismatch");
    }
}