Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 95 additions & 0 deletions packages/api-client/src/schema.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1993,6 +1993,16 @@ export interface components {
sourceJobIds: string[];
states: ("succeeded" | "failed" | "stopped" | "edited")[];
};
} | {
id?: string;
slug?: string;
/** @constant */
kind: "jobIdle";
/** @default true */
enabled: boolean;
config: {
sourceJobIds: string[];
};
})[];
skillIds?: string[];
};
Expand Down Expand Up @@ -4729,6 +4739,21 @@ export interface operations {
jobId: string;
createdAt: string;
updatedAt: string;
} | {
/** Format: uuid */
id: string;
slug: string;
/** @constant */
kind: "jobIdle";
/** @default true */
enabled: boolean;
config: {
sourceJobIds: string[];
};
/** Format: uuid */
jobId: string;
createdAt: string;
updatedAt: string;
})[];
skills?: {
/** Format: uuid */
Expand Down Expand Up @@ -4904,6 +4929,21 @@ export interface operations {
jobId: string;
createdAt: string;
updatedAt: string;
} | {
/** Format: uuid */
id: string;
slug: string;
/** @constant */
kind: "jobIdle";
/** @default true */
enabled: boolean;
config: {
sourceJobIds: string[];
};
/** Format: uuid */
jobId: string;
createdAt: string;
updatedAt: string;
})[];
skills?: {
/** Format: uuid */
Expand Down Expand Up @@ -5082,6 +5122,21 @@ export interface operations {
jobId: string;
createdAt: string;
updatedAt: string;
} | {
/** Format: uuid */
id: string;
slug: string;
/** @constant */
kind: "jobIdle";
/** @default true */
enabled: boolean;
config: {
sourceJobIds: string[];
};
/** Format: uuid */
jobId: string;
createdAt: string;
updatedAt: string;
})[];
skills?: {
/** Format: uuid */
Expand Down Expand Up @@ -5432,6 +5487,21 @@ export interface operations {
jobId: string;
createdAt: string;
updatedAt: string;
} | {
/** Format: uuid */
id: string;
slug: string;
/** @constant */
kind: "jobIdle";
/** @default true */
enabled: boolean;
config: {
sourceJobIds: string[];
};
/** Format: uuid */
jobId: string;
createdAt: string;
updatedAt: string;
})[];
skills?: {
/** Format: uuid */
Expand Down Expand Up @@ -5565,6 +5635,16 @@ export interface operations {
sourceJobIds: string[];
states: ("succeeded" | "failed" | "stopped" | "edited")[];
};
} | {
id?: string;
slug?: string;
/** @constant */
kind: "jobIdle";
/** @default true */
enabled?: boolean;
config: {
sourceJobIds: string[];
};
})[];
skillIds?: string[];
/** Format: date-time */
Expand Down Expand Up @@ -5715,6 +5795,21 @@ export interface operations {
jobId: string;
createdAt: string;
updatedAt: string;
} | {
/** Format: uuid */
id: string;
slug: string;
/** @constant */
kind: "jobIdle";
/** @default true */
enabled: boolean;
config: {
sourceJobIds: string[];
};
/** Format: uuid */
jobId: string;
createdAt: string;
updatedAt: string;
})[];
skills?: {
/** Format: uuid */
Expand Down
4 changes: 4 additions & 0 deletions packages/api/migrations/0023_job_idle_trigger.sql
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
ALTER TABLE app.triggers DROP CONSTRAINT triggers_kind_check;
ALTER TABLE app.triggers ADD CONSTRAINT triggers_kind_check CHECK (kind IN ('manual','schedule','webhook','jobLifecycle','jobIdle'));
ALTER TABLE app.invocations DROP CONSTRAINT invocations_source_check;
ALTER TABLE app.invocations ADD CONSTRAINT invocations_source_check CHECK (source IN ('manual','schedule','webhook','jobLifecycle','jobIdle'));
1 change: 1 addition & 0 deletions packages/api/src/flow-schemas.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ export const triggerSchema = z.discriminatedUnion("kind", [
z.object({ id: z.string().min(1).optional(), slug: z.string().regex(/^trigger-[1-9][0-9]*$/).optional(), kind: z.literal("jobLifecycle"), enabled: z.boolean().default(true), config: z.object({
sourceJobIds: z.array(z.string().min(1)).min(1), states: z.array(z.enum(["succeeded", "failed", "stopped", "edited"])).min(1),
}).strict() }).strict(),
z.object({ id: z.string().min(1).optional(), slug: z.string().regex(/^trigger-[1-9][0-9]*$/).optional(), kind: z.literal("jobIdle"), enabled: z.boolean().default(true), config: z.object({ sourceJobIds: z.array(z.string().min(1)).min(1) }).strict() }).strict(),
]);
export const jobInputSchema = z.object({
projectId: z.guid().optional(),
Expand Down
1 change: 1 addition & 0 deletions packages/api/src/job-contracts.ts
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ export const triggerResponse = z.discriminatedUnion("kind", [
triggerSchema.options[1].extend(triggerIdentity),
triggerSchema.options[2].extend({ ...triggerIdentity, config: webhookResponse }),
triggerSchema.options[3].extend(triggerIdentity),
triggerSchema.options[4].extend(triggerIdentity),
]);
export const jobResponse = z.object({
id: z.guid(), projectId: z.guid().optional(), name: z.string(), slug: z.string(), enabled: z.boolean(),
Expand Down
4 changes: 2 additions & 2 deletions packages/api/src/job-domain.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import Mustache from "mustache";
import { renderTextTemplate } from "./text-template";

export type TriggerKind = "manual" | "schedule" | "webhook" | "jobLifecycle";
export type TriggerKind = "manual" | "schedule" | "webhook" | "jobLifecycle" | "jobIdle";
export type InvocationSource = "manual" | TriggerKind;
export type JobRunState = "queued" | "reserved" | "running" | "succeeded" | "failed" | "stopped";

Expand Down Expand Up @@ -129,7 +129,7 @@ export class InvocationService {

async invoke(jobId: string, input: InvocationRequest): Promise<InvocationResult> {
if (!input.claimKey?.trim()) throw new InvocationError("invalid_invocation", "An invocation claim key is required.");
if (!(["manual", "schedule", "webhook", "jobLifecycle"] as const).includes(input.source)) throw new InvocationError("invalid_invocation", "Invocation source is invalid.");
if (!(["manual", "schedule", "webhook", "jobLifecycle", "jobIdle"] as const).includes(input.source)) throw new InvocationError("invalid_invocation", "Invocation source is invalid.");
const job = await this.repository.getJob(jobId);
if (!job) throw new InvocationError("job_not_found", "Job not found.");
if (!job.enabled) throw new InvocationError("job_disabled", "Job is disabled.");
Expand Down
29 changes: 28 additions & 1 deletion packages/api/src/postgres/automation-scheduler.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ function skipFullQueue(error: unknown): void {
/** Claims durable automatic signals in PostgreSQL. The alarm DO is only its clock. */
export class AutomationScheduler {
constructor(private database: Database, private env: Env) {}
async process(): Promise<void> { await this.githubVerifications(); await this.schedules(); await this.jobEdits(); await this.lifecycle(); }
async process(): Promise<void> { await this.githubVerifications(); await this.schedules(); await this.jobEdits(); await this.lifecycle(); await this.jobIdle(); }
private repository(tenantId: string) { return new PostgresJobRepository(this.database, tenantId, value => encrypt(value, this.env.CREDENTIAL_ENCRYPTION_KEY), value => decrypt(value, this.env.CREDENTIAL_ENCRYPTION_KEY)); }
private async schedules() {
for (let count = 0; count < 50; count++) {
Expand Down Expand Up @@ -73,6 +73,33 @@ export class AutomationScheduler {
}
}

private async jobIdle() {
const rows = await this.database.pool.query<any>(`SELECT t.tenant_id,t.id trigger_id,t.job_id,t.slug,r.id source_run_id,r.job_id source_job_id,r.updated_at
FROM app.triggers t JOIN app.jobs j ON j.tenant_id=t.tenant_id AND j.id=t.job_id
CROSS JOIN LATERAL jsonb_array_elements_text(t.config->'sourceJobIds') source_job(value)
JOIN LATERAL (SELECT id,job_id,updated_at FROM app.runs WHERE tenant_id=t.tenant_id AND job_id=source_job.value::uuid AND state IN ('succeeded','failed','stopped') ORDER BY updated_at DESC,id DESC LIMIT 1) r ON true
WHERE t.kind='jobIdle' AND t.enabled AND t.removed_at IS NULL AND j.enabled
AND r.updated_at > t.updated_at
AND NOT EXISTS (
SELECT 1 FROM jsonb_array_elements_text(t.config->'sourceJobIds') watched(value)
JOIN app.runs active ON active.tenant_id=t.tenant_id AND active.job_id=watched.value::uuid
WHERE active.state IN ('queued','reserved','starting','running','blocked','stopping')
)
AND NOT EXISTS (
SELECT 1 FROM jsonb_array_elements_text(t.config->'sourceJobIds') watched(value)
JOIN app.runs newer ON newer.tenant_id=t.tenant_id AND newer.job_id=watched.value::uuid
WHERE newer.state IN ('succeeded','failed','stopped')
AND (newer.updated_at,newer.id) > (r.updated_at,r.id)
)
AND NOT EXISTS (SELECT 1 FROM app.lifecycle_deliveries d WHERE d.tenant_id=t.tenant_id AND d.trigger_id=t.id AND d.source_run_id=r.id AND d.terminal_state='runsZero')
ORDER BY r.updated_at LIMIT 100`);
for (const row of rows.rows) {
const completedAt = row.updated_at.toISOString(), context = { source_job_id: row.source_job_id, event: "runsZero", completed_at: completedAt };
await new InvocationService(this.repository(row.tenant_id), value => encrypt(value, this.env.CREDENTIAL_ENCRYPTION_KEY)).invoke(row.job_id, { source: "jobIdle", triggerId: row.trigger_id, claimKey: `job-idle:${row.trigger_id}:${row.source_run_id}`, context: { [row.slug]: context }, occurrence: { occurredAt: completedAt, externalId: row.source_run_id, metadata: context } }).catch(skipFullQueue);
await this.database.pool.query("INSERT INTO app.lifecycle_deliveries(tenant_id,trigger_id,source_run_id,terminal_state) VALUES ($1,$2,$3,'runsZero') ON CONFLICT DO NOTHING", [row.tenant_id, row.trigger_id, row.source_run_id]);
}
}

private async githubVerifications() {
const pending = await this.database.pool.query<any>(`SELECT p.*,t.slug trigger_slug,t.config trigger_config,t.enabled trigger_enabled,t.removed_at,j.enabled job_enabled,g.state installation_state FROM app.pending_verifications p JOIN app.triggers t ON t.tenant_id=p.tenant_id AND t.id=p.trigger_id JOIN app.jobs j ON j.tenant_id=p.tenant_id AND j.id=p.job_id LEFT JOIN app.github_installations g ON g.tenant_id=p.tenant_id AND g.installation_id=p.installation_id WHERE p.next_attempt_at<=now() ORDER BY p.next_attempt_at LIMIT 20`);
for (const row of pending.rows) {
Expand Down
4 changes: 2 additions & 2 deletions packages/api/src/postgres/schema.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,14 +54,14 @@ export const skills = app.table("skills", {
export const jobSkills = app.table("job_skills", { tenantId: uuid("tenant_id").notNull(), jobId: uuid("job_id").notNull(), skillId: uuid("skill_id").notNull(), position: integer("position").notNull() }, t => [primaryKey({ columns: [t.tenantId, t.jobId, t.skillId] }), unique().on(t.tenantId, t.jobId, t.position), foreignKey({ columns: [t.tenantId, t.jobId], foreignColumns: [jobs.tenantId, jobs.id] }).onDelete("cascade"), foreignKey({ columns: [t.tenantId, t.skillId], foreignColumns: [skills.tenantId, skills.id] }).onDelete("restrict")]);
export const triggers = app.table("triggers", {
tenantId: uuid("tenant_id").notNull(), id: uuid("id").notNull(), jobId: uuid("job_id").notNull(), kind: text("kind").$type<TriggerKind>().notNull(), slug: text("slug").notNull(), enabled: boolean("enabled").notNull().default(true), config: jsonb("config").$type<Record<string, unknown>>().notNull().default({}), position: integer("position").notNull().default(0), createdAt: createdAt(), updatedAt: updatedAt(), removedAt: timestamp("removed_at", { withTimezone: true }),
}, t => [primaryKey({ columns: [t.tenantId, t.id] }), uniqueIndex("triggers_active_slug").on(t.tenantId, t.jobId, t.slug).where(sql`${t.removedAt} is null`), foreignKey({ columns: [t.tenantId, t.jobId], foreignColumns: [jobs.tenantId, jobs.id] }).onDelete("cascade"), check("no_webhook_handler_code", sql`NOT (${t.config} ? 'handlerCode')`), check("triggers_kind_check", sql`${t.kind} in ('manual','schedule','webhook','jobLifecycle')`)]);
}, t => [primaryKey({ columns: [t.tenantId, t.id] }), uniqueIndex("triggers_active_slug").on(t.tenantId, t.jobId, t.slug).where(sql`${t.removedAt} is null`), foreignKey({ columns: [t.tenantId, t.jobId], foreignColumns: [jobs.tenantId, jobs.id] }).onDelete("cascade"), check("no_webhook_handler_code", sql`NOT (${t.config} ? 'handlerCode')`), check("triggers_kind_check", sql`${t.kind} in ('manual','schedule','webhook','jobLifecycle','jobIdle')`)]);
// One-time cutover state; no legacy execution or configuration is retained.
export const webhookConditionsCutover = app.table("webhook_conditions_cutover", {
tenantId: uuid("tenant_id").notNull(), triggerId: uuid("trigger_id").notNull(), wasEnabled: boolean("was_enabled").notNull(), conditions: jsonb("conditions").$type<Record<string, unknown>>().notNull(), restoredAt: timestamp("restored_at", { withTimezone: true }),
}, t => [primaryKey({ columns: [t.tenantId, t.triggerId] }), foreignKey({ columns: [t.tenantId, t.triggerId], foreignColumns: [triggers.tenantId, triggers.id] }).onDelete("cascade")]);
export const invocations = app.table("invocations", {
tenantId: uuid("tenant_id").notNull(), id: uuid("id").notNull(), jobId: uuid("job_id").notNull(), source: text("source").$type<InvocationSource>().notNull(), claimKey: text("claim_key").notNull(), triggerId: uuid("trigger_id").notNull(), context: jsonb("context").$type<Record<string, unknown>>().notNull(), occurrence: jsonb("occurrence").$type<InvocationRequest["occurrence"]>(), createdAt: createdAt(),
}, t => [primaryKey({ columns: [t.tenantId, t.id] }), unique().on(t.tenantId, t.jobId, t.claimKey), foreignKey({ columns: [t.tenantId, t.jobId], foreignColumns: [jobs.tenantId, jobs.id] }).onDelete("cascade"), foreignKey({ columns: [t.tenantId, t.triggerId], foreignColumns: [triggers.tenantId, triggers.id] }), check("invocations_source_check", sql`${t.source} in ('manual','schedule','webhook','jobLifecycle')`)]);
}, t => [primaryKey({ columns: [t.tenantId, t.id] }), unique().on(t.tenantId, t.jobId, t.claimKey), foreignKey({ columns: [t.tenantId, t.jobId], foreignColumns: [jobs.tenantId, jobs.id] }).onDelete("cascade"), foreignKey({ columns: [t.tenantId, t.triggerId], foreignColumns: [triggers.tenantId, triggers.id] }), check("invocations_source_check", sql`${t.source} in ('manual','schedule','webhook','jobLifecycle','jobIdle')`)]);
export const jobRuns = app.table("job_runs", {
tenantId: uuid("tenant_id").notNull(), id: uuid("id").notNull(), jobId: uuid("job_id").notNull(), invocationId: uuid("invocation_id").notNull(), state: text("state").$type<"queued" | "reserved" | "starting" | "running" | "blocked" | "stopping" | "succeeded" | "failed" | "stopped">().notNull(), encryptedPrompt: text("encrypted_prompt").notNull(), runName: text("run_name").notNull().default(""), launchLeaseExpiresAt: timestamp("launch_lease_expires_at", { withTimezone: true }), nextPollAt: timestamp("next_poll_at", { withTimezone: true }), createdAt: createdAt(), updatedAt: updatedAt(), startedAt: timestamp("started_at", { withTimezone: true }),
}, t => [primaryKey({ columns: [t.tenantId, t.id] }), unique().on(t.tenantId, t.invocationId), foreignKey({ columns: [t.tenantId, t.jobId], foreignColumns: [jobs.tenantId, jobs.id] }).onDelete("cascade"), foreignKey({ columns: [t.tenantId, t.invocationId], foreignColumns: [invocations.tenantId, invocations.id] }).onDelete("cascade"), index("job_runs_queue").on(t.state, t.createdAt).where(sql`${t.state}='queued'`), index("job_runs_poll_due").on(t.nextPollAt).where(sql`${t.state} in ('running','blocked','stopping')`), index("job_runs_launch_lease").on(t.launchLeaseExpiresAt).where(sql`${t.state}='starting'`)]);
Expand Down
5 changes: 5 additions & 0 deletions packages/api/src/trigger-context.ts
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,11 @@ export const triggerContextCatalog: Record<string, ContextPath[]> = {
{ path: "logs", type: "array", description: "Sanitized console log records." },
{ path: "exceptions", type: "array", description: "Sanitized exception records." },
],
jobIdle: [
{ path: "source_job_id", type: "string", description: "Job whose runs have all completed." },
{ path: "event", type: "string", description: "Event name: runsZero." },
{ path: "completed_at", type: "string", description: "Completion timestamp of the last run (ISO 8601)." },
],
jobLifecycle: [
{ path: "event", type: "string", description: "Event name: edited for a job edit." },
{ path: "edited_at", type: "string", description: "Job save timestamp for an edited event (ISO 8601)." },
Expand Down
6 changes: 6 additions & 0 deletions packages/api/test/job-edited.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,12 @@ it("accepts edited alongside terminal events and rejects unknown lifecycle event
expect(triggerSchema.safeParse({ ...trigger, config: { ...trigger.config, states: ["unknown"] } }).success).toBe(false);
});

it("accepts an all-runs-complete trigger with one or more source jobs", () => {
const trigger = { kind: "jobIdle", config: { sourceJobIds: ["source"] } };
expect(triggerSchema.parse(trigger).config).toEqual(trigger.config);
expect(triggerSchema.safeParse({ ...trigger, config: { sourceJobIds: [] } }).success).toBe(false);
});

// Use a dedicated migrated database; never a shared application database.
describe.skipIf(!process.env.TEST_DATABASE_URL)("job edit delivery (PostgreSQL)", () => {
const database = new Database({ connectionString: process.env.TEST_DATABASE_URL });
Expand Down
Loading
Loading