Skip to content

refactor(worker): move job handler data access into business services #1101

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
realcodesiman wants to merge 3 commits into
base: main
Choose a base branch
from
Open

refactor(worker): move job handler data access into business services #1101

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
815146f
refactor(worker): move job handler data access into business services
realcodesiman Sep 6, 2026
8095f21
fix(worker): align tests and cleanup with data-access service refactor
realcodesiman Sep 12, 2026
c9666db
test(worker,database,business): add PR #1101 review follow-up coverage
realcodesiman Sep 13, 2026
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
78 changes: 30 additions & 48 deletions apps/worker/__tests__/action-executor-contact-inbox-attribution.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,14 +12,12 @@ import { beforeEach, describe, expect, test, vi } from "vitest"
// ---------------------------------------------------------------------------

const mocks = vi.hoisted(() => ({
conversationFindFirst: vi.fn(),
tagFindMany: vi.fn(),
flowFindFirst: vi.fn(),
workspaceMemberFindFirst: vi.fn(),
inboxTeamFindFirst: vi.fn(),
findLatestCreatedByContact: vi.fn(),
attachExistingToContactForTrigger: vi.fn(),
detachFromContactForTrigger: vi.fn(),
findActiveById: vi.fn(),
findByIdForContact: vi.fn(),
findMostRecentByContact: vi.fn(),
insertReturning: vi.fn(),
enqueueEvent: vi.fn(),
buildSourceKey: vi.fn(),
setValues: vi.fn(),
Expand All @@ -36,45 +34,7 @@ const mocks = vi.hoisted(() => ({
getSpreadsheetRow: vi.fn(),
}))

vi.mock("@chatbotx.io/database/client", () => ({
db: {
query: {
conversationModel: {
findFirst: (...args: unknown[]) => mocks.conversationFindFirst(...args),
},
tagModel: {
findMany: (...args: unknown[]) => mocks.tagFindMany(...args),
},
flowModel: {
findFirst: (...args: unknown[]) => mocks.flowFindFirst(...args),
},
workspaceMemberModel: {
findFirst: (...args: unknown[]) =>
mocks.workspaceMemberFindFirst(...args),
},
inboxTeamModel: {
findFirst: (...args: unknown[]) => mocks.inboxTeamFindFirst(...args),
},
},
insert: () => ({
values: () => ({
onConflictDoNothing: () => ({
returning: (...args: unknown[]) => mocks.insertReturning(...args),
}),
}),
}),
delete: () => ({ where: vi.fn() }),
},
and: (...args: unknown[]) => ({ and: args }),
eq: (col: unknown, val: unknown) => ({ eq: [col, val] }),
inArray: (col: unknown, vals: unknown) => ({ inArray: [col, vals] }),
}))

vi.mock("@chatbotx.io/database/schema", () => ({
contactsToTagsModel: {
contactId: "contactsToTagsModel.contactId",
tagId: "contactsToTagsModel.tagId",
},
metaCapiEventChannelSchema: {
safeParse: (value: unknown) =>
value === "messenger" || value === "instagram" || value === "whatsapp"
Expand All @@ -99,12 +59,23 @@ vi.mock("@chatbotx.io/business", () => ({
mocks.deleteByCustomFieldId(...args),
},
conversationService: {
findLatestCreatedByContact: (...args: unknown[]) =>
mocks.findLatestCreatedByContact(...args),
updateArchived: (...args: unknown[]) => mocks.updateArchived(...args),
updateAssignment: (...args: unknown[]) => mocks.updateAssignment(...args),
assignOneOrSkip: (...args: unknown[]) => mocks.assignOneOrSkip(...args),
disableBotState: (...args: unknown[]) => mocks.disableBotState(...args),
enableBotState: (...args: unknown[]) => mocks.enableBotState(...args),
},
tagService: {
attachExistingToContactForTrigger: (...args: unknown[]) =>
mocks.attachExistingToContactForTrigger(...args),
detachFromContactForTrigger: (...args: unknown[]) =>
mocks.detachFromContactForTrigger(...args),
},
flowService: {
findActiveById: (...args: unknown[]) => mocks.findActiveById(...args),
},
tagSyncService: {
enqueueAttach: (...args: unknown[]) => mocks.enqueueAttach(...args),
enqueueDetach: (...args: unknown[]) => mocks.enqueueDetach(...args),
Expand All @@ -123,6 +94,16 @@ vi.mock("@chatbotx.io/events/context", () => ({
webhookChannelOrigin: vi.fn(() => "webhook"),
}))

// `capi-input-error.ts` (imported transitively for the sendMetaCapiEvent
// branch) pulls `logProviderError` from this separate package subpath, which
// is not covered by the `@chatbotx.io/business` mock above (subpath exports
// are independent module specifiers). Left unmocked, it loads the real
// `@chatbotx.io/database/client` and relations graph against the partial
// schema mock below and crashes at import time.
vi.mock("@chatbotx.io/business/error-log", () => ({
logProviderError: vi.fn(),
}))

vi.mock("@chatbotx.io/logger", () => ({
default: { warn: vi.fn(), error: vi.fn(), info: vi.fn() },
getChildLogger: () => ({
Expand Down Expand Up @@ -179,7 +160,7 @@ const MESSENGER_INBOX = {
describe("ActionExecutor — per-integration contact inbox attribution", () => {
beforeEach(() => {
vi.clearAllMocks()
mocks.conversationFindFirst.mockResolvedValue({
mocks.findLatestCreatedByContact.mockResolvedValue({
id: "conv-1",
contactId: "contact-1",
workspaceId: "ws-1",
Expand All @@ -194,7 +175,7 @@ describe("ActionExecutor — per-integration contact inbox attribution", () => {
mocks.findByIdForContact.mockResolvedValue(WHATSAPP_INBOX)
mocks.findMostRecentByContact.mockResolvedValue(MESSENGER_INBOX)
mocks.buildSourceKey.mockReturnValue("source-key")
mocks.flowFindFirst.mockResolvedValue({
mocks.findActiveById.mockResolvedValue({
id: "flow-1",
currentVersionId: "fv-1",
})
Expand Down Expand Up @@ -328,8 +309,9 @@ describe("ActionExecutor — per-integration contact inbox attribution", () => {
mocks.findMostRecentByContact.mockRejectedValue(
new Error("resolver should never be called for this branch"),
)
mocks.tagFindMany.mockResolvedValue([{ id: "tag-1" }])
mocks.insertReturning.mockResolvedValue([{ tagId: "tag-1" }])
mocks.attachExistingToContactForTrigger.mockResolvedValue([
{ tagId: "tag-1" },
])
mocks.assignOneOrSkip.mockResolvedValue(undefined)
})

Expand Down
32 changes: 11 additions & 21 deletions apps/worker/__tests__/condition-evaluator.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,20 +3,14 @@ import type { WorkspaceModel } from "@chatbotx.io/database/types"
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"
import type { ConditionEvaluationContext } from "../src/trigger/types"

const { contactCustomFieldFindFirst, customFieldFindFirst } = vi.hoisted(
() => ({
contactCustomFieldFindFirst: vi.fn(),
customFieldFindFirst: vi.fn(),
}),
)

vi.mock("@chatbotx.io/database/client", () => ({
db: {
query: {
contactCustomFieldModel: { findFirst: contactCustomFieldFindFirst },
customFieldModel: { findFirst: customFieldFindFirst },
},
},
const { contactCustomFieldFindValue, customFieldFindBy } = vi.hoisted(() => ({
contactCustomFieldFindValue: vi.fn(),
customFieldFindBy: vi.fn(),
}))

vi.mock("@chatbotx.io/business", () => ({
contactCustomFieldService: { findValue: contactCustomFieldFindValue },
customFieldService: { findBy: customFieldFindBy },
}))

import { ConditionEvaluator } from "../src/trigger/services/condition-evaluator"
Expand Down Expand Up @@ -110,9 +104,7 @@ describe("ConditionEvaluator dateTimeBasedTrigger timezone", () => {
// 14:00 UTC is 21:00 in Asia/Ho_Chi_Minh (+7), so an `at: "21"` condition
// only fires when the hour-of-day is resolved in the +7 zone, never in UTC.
vi.setSystemTime(new Date("2026-07-11T14:00:00.000Z"))
contactCustomFieldFindFirst.mockResolvedValue({
value: "2026-07-11T02:00:00.000Z",
})
contactCustomFieldFindValue.mockResolvedValue("2026-07-11T02:00:00.000Z")
})

afterEach(() => {
Expand Down Expand Up @@ -178,9 +170,7 @@ describe("ConditionEvaluator dateTimeBasedTrigger date-type anchor", () => {
// zone. The VN date 2026-07-11 is stored as 2026-07-11T00:00:00+07:00.
// The anchor is the START of the day (hour 0), never the legacy end-of-day
// (hour 23).
contactCustomFieldFindFirst.mockResolvedValue({
value: "2026-07-11T00:00:00+07:00",
})
contactCustomFieldFindValue.mockResolvedValue("2026-07-11T00:00:00+07:00")
})

afterEach(() => {
Expand Down Expand Up @@ -249,7 +239,7 @@ describe("ConditionEvaluator customFieldValueChanged operator vocabulary", () =>
fieldType: string,
newValue: unknown,
): ConditionEvaluationContext => {
customFieldFindFirst.mockResolvedValue({ type: fieldType })
customFieldFindBy.mockResolvedValue({ type: fieldType })
return buildContext(
{
type: triggerEventTypes.enum.customFieldValueChanged,
Expand Down
146 changes: 76 additions & 70 deletions apps/worker/__tests__/datetime-trigger-evaluator.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,34 +3,26 @@ import { beforeEach, describe, expect, test, vi } from "vitest"

const {
actionExecute,
insertTriggerExecution,
listActiveWithConditionsPage,
listContactCustomFieldsForDateTimeSweep,
listContactCustomFieldsForDateTimeSweepContacts,
triggerExecutionFindMany,
triggerFindMany,
listExecutedPairs,
recordExecution,
} = vi.hoisted(() => ({
actionExecute: vi.fn(),
insertTriggerExecution: vi.fn(),
listActiveWithConditionsPage: vi.fn(),
listContactCustomFieldsForDateTimeSweep: vi.fn(),
listContactCustomFieldsForDateTimeSweepContacts: vi.fn(),
triggerExecutionFindMany: vi.fn(),
triggerFindMany: vi.fn(),
listExecutedPairs: vi.fn(),
recordExecution: vi.fn(),
}))

vi.mock("@chatbotx.io/database/client", () => ({
db: {
execute: vi.fn(),
insert: insertTriggerExecution,
query: {
triggerExecutionModel: {
findMany: triggerExecutionFindMany,
},
triggerModel: {
findMany: triggerFindMany,
},
},
vi.mock("@chatbotx.io/business", () => ({
triggerService: {
listActiveWithConditionsPage,
listExecutedPairs,
recordExecution,
},
sql: vi.fn(),
}))

vi.mock("@chatbotx.io/database/repositories", () => ({
Expand Down Expand Up @@ -112,12 +104,8 @@ describe("evaluateDateTimeTriggers", () => {
beforeEach(() => {
vi.clearAllMocks()
actionExecute.mockResolvedValue(undefined)
triggerExecutionFindMany.mockResolvedValue([])
insertTriggerExecution.mockReturnValue({
values: vi.fn().mockReturnValue({
onConflictDoNothing: vi.fn().mockResolvedValue(undefined),
}),
})
listExecutedPairs.mockResolvedValue([])
recordExecution.mockResolvedValue(undefined)
redis.get.mockResolvedValue(null)
redis.set.mockResolvedValue("OK")
redis.setex.mockResolvedValue("OK")
Expand All @@ -137,14 +125,20 @@ describe("evaluateDateTimeTriggers", () => {
}),
),
]
triggerFindMany
.mockResolvedValueOnce(firstTriggerChunk)
.mockResolvedValueOnce([
triggerRow({
id: "trigger-101",
conditions: [dateTimeCondition("field-2")],
}),
])
listActiveWithConditionsPage
.mockResolvedValueOnce({
triggers: firstTriggerChunk,
nextCursor: "trigger-100",
})
.mockResolvedValueOnce({
triggers: [
triggerRow({
id: "trigger-101",
conditions: [dateTimeCondition("field-2")],
}),
],
nextCursor: undefined,
})
listContactCustomFieldsForDateTimeSweep
.mockResolvedValueOnce({
rows: [
Expand Down Expand Up @@ -220,15 +214,18 @@ describe("evaluateDateTimeTriggers", () => {
})

test("waits for all datetime conditions before executing a trigger across cursor pages", async () => {
triggerFindMany.mockResolvedValueOnce([
triggerRow({
id: "trigger-001",
conditions: [
dateTimeCondition("field-1"),
dateTimeCondition("field-2"),
],
}),
])
listActiveWithConditionsPage.mockResolvedValueOnce({
triggers: [
triggerRow({
id: "trigger-001",
conditions: [
dateTimeCondition("field-1"),
dateTimeCondition("field-2"),
],
}),
],
nextCursor: undefined,
})
listContactCustomFieldsForDateTimeSweep
.mockResolvedValueOnce({
rows: [
Expand Down Expand Up @@ -287,15 +284,18 @@ describe("evaluateDateTimeTriggers", () => {
})

test("does not execute a multi-condition trigger when only one datetime condition is present", async () => {
triggerFindMany.mockResolvedValueOnce([
triggerRow({
id: "trigger-001",
conditions: [
dateTimeCondition("field-1"),
dateTimeCondition("field-2"),
],
}),
])
listActiveWithConditionsPage.mockResolvedValueOnce({
triggers: [
triggerRow({
id: "trigger-001",
conditions: [
dateTimeCondition("field-1"),
dateTimeCondition("field-2"),
],
}),
],
nextCursor: undefined,
})
listContactCustomFieldsForDateTimeSweep.mockResolvedValueOnce({
rows: [
contactCustomFieldRow({
Expand Down Expand Up @@ -324,18 +324,21 @@ describe("evaluateDateTimeTriggers", () => {
// Workspace is UTC, but the condition was saved in Asia/Ho_Chi_Minh (+7).
// 14:00 UTC is 21:00 in +7, so `at: "21"` only fires when the condition's
// own zone is honored — a UTC resolution would land on hour 14 and miss.
triggerFindMany.mockResolvedValueOnce([
triggerRow({
id: "trigger-001",
timezone: "UTC",
conditions: [
dateTimeCondition("field-1", {
at: "21",
timezone: "Asia/Ho_Chi_Minh",
}),
],
}),
])
listActiveWithConditionsPage.mockResolvedValueOnce({
triggers: [
triggerRow({
id: "trigger-001",
timezone: "UTC",
conditions: [
dateTimeCondition("field-1", {
at: "21",
timezone: "Asia/Ho_Chi_Minh",
}),
],
}),
],
nextCursor: undefined,
})
listContactCustomFieldsForDateTimeSweep.mockResolvedValueOnce({
rows: [
contactCustomFieldRow({
Expand Down Expand Up @@ -367,13 +370,16 @@ describe("evaluateDateTimeTriggers", () => {
test("falls back to the workspace timezone for legacy conditions with no captured zone", async () => {
// The condition predates timezone capture (no zone stored), so day
// boundaries and hour-of-day must resolve in the workspace zone (+7).
triggerFindMany.mockResolvedValueOnce([
triggerRow({
id: "trigger-001",
timezone: "Asia/Ho_Chi_Minh",
conditions: [dateTimeCondition("field-1", { at: "21" })],
}),
])
listActiveWithConditionsPage.mockResolvedValueOnce({
triggers: [
triggerRow({
id: "trigger-001",
timezone: "Asia/Ho_Chi_Minh",
conditions: [dateTimeCondition("field-1", { at: "21" })],
}),
],
nextCursor: undefined,
})
listContactCustomFieldsForDateTimeSweep.mockResolvedValueOnce({
rows: [
contactCustomFieldRow({
Expand Down
Loading
Loading