ZacharyZhang-NY
b1d1fbc54f
Before this commit `createDocumentWithLifecycle` enqueued ledger rows
inside the save transaction but consumed them from an in-memory
`pendingSideEffects` array straight after commit. A crash between the
commit and the drain stranded rows as `pending` forever; a crash mid
handler left them as `running`; and there was no retry budget.
The new `DocumentLifecycleProcessor` is the single drainer. It:
- claims work via `claimPendingBatch`, an atomic `UPDATE … RETURNING`
that flips rows from `pending` to `running` while incrementing
`attempts`, so two drainers cannot race onto the same row;
- dispatches each row through a handler registered by `hookType`,
parsing the frozen `payload_json` back into the call arguments;
- records `succeeded` / `failed` / `retryable` outcomes through the
ledger, so a transient failure stays on the queue until the attempt
cap expires, after which it is marked `failed` for operators;
- reclaims rows stuck in `running` past `staleRunningMs` back to
`pending` on every sweep tick plus at startup, so a crashed process
never leaves a row orphaned;
- runs a periodic sweep so every document — including those whose
immediate post-commit drain failed — is eventually advanced.
`createDocumentWithLifecycle` now only enqueues and then calls
`lifecycleProcessor.drainForDocument` for the immediate happy-path pass.
Handlers live in `document-lifecycle-handlers.ts` so the processor stays
free of service-graph knowledge. The processor is constructed in
`bootstrap-application`, started there, and stopped on `before-quit`.
76 lines
2.6 KiB
TypeScript
76 lines
2.6 KiB
TypeScript
/**
|
|
* Registers the concrete lifecycle hook handlers against the processor.
|
|
* Handler side is kept thin — read the payload, call into the service
|
|
* graph, translate throws into a `retryable` outcome so the processor's
|
|
* attempt cap decides when a transient failure becomes permanent.
|
|
*/
|
|
|
|
import type { AutomationEventContext } from "./automation/types";
|
|
import type {
|
|
DocumentLifecycleProcessor,
|
|
LifecycleHookContext,
|
|
LifecycleHookOutcome,
|
|
} from "./document-lifecycle-processor";
|
|
|
|
export interface LifecycleHandlerDeps {
|
|
dispatchAutomation: (trigger: string, ctx: AutomationEventContext) => void;
|
|
wakeAgentIfPossible: (
|
|
agentId: string | null | undefined,
|
|
companyId: string,
|
|
trigger: string,
|
|
) => string | null;
|
|
}
|
|
|
|
export function registerLifecycleHandlers(
|
|
processor: DocumentLifecycleProcessor,
|
|
deps: LifecycleHandlerDeps,
|
|
): void {
|
|
processor.register("automation_dispatch", automationDispatchHandler(deps));
|
|
processor.register("reviewer_wake", reviewerWakeHandler(deps));
|
|
}
|
|
|
|
function automationDispatchHandler(deps: LifecycleHandlerDeps) {
|
|
return (payload: Record<string, unknown>, context: LifecycleHookContext): LifecycleHookOutcome => {
|
|
const trigger = typeof payload.trigger === "string" ? payload.trigger : "document_created";
|
|
const automationContext: AutomationEventContext = {
|
|
companyId: context.companyId,
|
|
agentId: stringOrUndefined(payload.agentId),
|
|
projectId: stringOrUndefined(payload.projectId),
|
|
};
|
|
try {
|
|
deps.dispatchAutomation(trigger, automationContext);
|
|
return { kind: "succeeded" };
|
|
} catch (error) {
|
|
return { kind: "retryable", reason: describeError(error) };
|
|
}
|
|
};
|
|
}
|
|
|
|
function reviewerWakeHandler(deps: LifecycleHandlerDeps) {
|
|
return (payload: Record<string, unknown>, context: LifecycleHookContext): LifecycleHookOutcome => {
|
|
const reviewerAgentId = stringOrNull(payload.reviewerAgentId);
|
|
if (!reviewerAgentId) {
|
|
return { kind: "failed", reason: "payload missing reviewerAgentId" };
|
|
}
|
|
try {
|
|
deps.wakeAgentIfPossible(reviewerAgentId, context.companyId, "assignment");
|
|
return { kind: "succeeded" };
|
|
} catch (error) {
|
|
return { kind: "retryable", reason: describeError(error) };
|
|
}
|
|
};
|
|
}
|
|
|
|
function stringOrUndefined(value: unknown): string | undefined {
|
|
return typeof value === "string" && value.length > 0 ? value : undefined;
|
|
}
|
|
|
|
function stringOrNull(value: unknown): string | null {
|
|
return typeof value === "string" && value.length > 0 ? value : null;
|
|
}
|
|
|
|
function describeError(error: unknown): string {
|
|
if (error instanceof Error) return error.message;
|
|
return String(error);
|
|
}
|