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
38 changes: 38 additions & 0 deletions packages/worker/src/jobs/manager-client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,29 @@ import {
type JobView,
} from './types.ts'

export type JobManagerDebugStatus =
| 'missing_binding'
| 'idle'
| 'armed'
| 'out_of_sync'

export type JobManagerDebugState = {
bindingAvailable: boolean
status: JobManagerDebugStatus
storedUserId: string | null
alarmScheduledFor: string | null
nextRunnableJobId: string | null
nextRunnableRunAt: string | null
alarmInSync: boolean | null
}

type JobManagerRpc = {
syncAlarm: (payload: { userId: string }) => Promise<{
ok: true
userId: string
nextRunAt: string | null
}>
getDebugState: (payload: { userId: string }) => Promise<JobManagerDebugState>
runNow: (payload: {
userId: string
jobId: string
Expand Down Expand Up @@ -44,6 +61,27 @@ export async function syncJobManagerAlarm(input: { env: Env; userId: string }) {
})
}

export async function getJobManagerDebugState(input: {
env: Env
userId: string
}) {
const rpc = jobManagerRpc(input.env, input.userId)
if (!rpc) {
return {
bindingAvailable: false,
status: 'missing_binding' as const,
storedUserId: null,
alarmScheduledFor: null,
nextRunnableJobId: null,
nextRunnableRunAt: null,
alarmInSync: null,
}
}
return rpc.getDebugState({
userId: input.userId,
})
}

export async function runJobNowViaManager(input: {
env: Env
userId: string
Expand Down
41 changes: 41 additions & 0 deletions packages/worker/src/jobs/manager-do.node.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
import { expect, test } from 'vitest'
import { resolveJobManagerAlarmState } from './manager-state.ts'

test('resolveJobManagerAlarmState treats equivalent UTC formats as in sync', () => {
expect(
resolveJobManagerAlarmState({
alarmTimestamp: Date.parse('2026-04-20T18:30:00.000Z'),
nextRunnableRunAt: '2026-04-20T18:30:00Z',
}),
).toEqual({
alarmScheduledFor: '2026-04-20T18:30:00.000Z',
alarmInSync: true,
status: 'armed',
})
})

test('resolveJobManagerAlarmState marks mismatched alarm times as out of sync', () => {
expect(
resolveJobManagerAlarmState({
alarmTimestamp: Date.parse('2026-04-20T19:00:00.000Z'),
nextRunnableRunAt: '2026-04-20T18:30:00.000Z',
}),
).toEqual({
alarmScheduledFor: '2026-04-20T19:00:00.000Z',
alarmInSync: false,
status: 'out_of_sync',
})
})

test('resolveJobManagerAlarmState reports idle when no alarm or next runnable job exists', () => {
expect(
resolveJobManagerAlarmState({
alarmTimestamp: null,
nextRunnableRunAt: null,
}),
).toEqual({
alarmScheduledFor: null,
alarmInSync: true,
status: 'idle',
})
})
41 changes: 40 additions & 1 deletion packages/worker/src/jobs/manager-do.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,13 @@ import { type McpCallerContext } from '@kody-internal/shared/chat.ts'
import { DurableObject } from 'cloudflare:workers'
import { buildSentryOptions } from '#worker/sentry-options.ts'
import { getNextRunnableJob, runDueJobsForUser, runJobNow } from './service.ts'
import { type JobRepoCheckPolicy } from './types.ts'
import { resolveJobManagerAlarmState } from './manager-state.ts'
import {
type JobRepoCheckPolicy,
} from './types.ts'
import {
type JobManagerDebugState,
} from './manager-client.ts'

const userIdStorageKey = 'user-id'

Expand Down Expand Up @@ -34,6 +40,36 @@ class JobManagerBase extends DurableObject<Env> {
}
}

async getDebugState(input: { userId: string }): Promise<JobManagerDebugState> {
const userId = input.userId.trim()
if (!userId) {
throw new Error('Job manager requires a non-empty userId.')
}
const [storedUserId, alarmTimestamp, nextJob] = await Promise.all([
this.ctx.storage.get<string>(userIdStorageKey),
this.ctx.storage.getAlarm(),
getNextRunnableJob({
env: this.env,
userId,
}),
])
const nextRunnableRunAt = nextJob?.nextRunAt ?? null
const { alarmScheduledFor, alarmInSync, status } =
resolveJobManagerAlarmState({
alarmTimestamp,
nextRunnableRunAt,
})
return {
bindingAvailable: true,
status,
storedUserId: storedUserId ?? null,
alarmScheduledFor,
nextRunnableJobId: nextJob?.id ?? null,
nextRunnableRunAt,
alarmInSync,
}
}

async alarm(): Promise<void> {
const userId = await this.ctx.storage.get<string>(userIdStorageKey)
if (!userId) {
Expand Down Expand Up @@ -94,6 +130,9 @@ export function jobManagerRpc(env: Env, userId: string) {
userId: string
nextRunAt: string | null
}>
getDebugState: (payload: {
userId: string
}) => Promise<JobManagerDebugState>
runNow: (payload: {
userId: string
jobId: string
Expand Down
38 changes: 38 additions & 0 deletions packages/worker/src/jobs/manager-state.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
import { type JobManagerDebugState, type JobManagerDebugStatus } from './manager-client.ts'

export function resolveJobManagerAlarmState(input: {
alarmTimestamp: number | null
nextRunnableRunAt: string | null
}): Pick<JobManagerDebugState, 'alarmScheduledFor' | 'alarmInSync' | 'status'> {
const alarmScheduledFor =
typeof input.alarmTimestamp === 'number'
? new Date(input.alarmTimestamp).toISOString()
: null
const nextRunnableMs =
input.nextRunnableRunAt == null
? null
: new Date(input.nextRunnableRunAt).valueOf()
const alarmMs =
typeof input.alarmTimestamp === 'number' ? input.alarmTimestamp : null
const timestampsMatch =
nextRunnableMs != null &&
alarmMs != null &&
Number.isFinite(nextRunnableMs) &&
alarmMs === nextRunnableMs
const alarmInSync =
nextRunnableMs == null ? alarmMs == null : timestampsMatch
const status: JobManagerDebugStatus =
nextRunnableMs == null
? alarmMs == null
? 'idle'
: 'out_of_sync'
: timestampsMatch
? 'armed'
: 'out_of_sync'

return {
alarmScheduledFor,
alarmInSync,
status,
}
}
Loading
Loading