mirror of
https://github.com/hanzoai/insights.git
synced 2026-10-30 16:15:31 +00:00
zeekay
1b367e2b62
The tree was a debranded snapshot of upstream taken around 2026-02-20, and upstream has moved about five months since. This replaces it with upstream master, mechanically re-derived: bin/debrand does the rename, and the pipeline and squashed migrations are restored from the shipped tree. Not a merge. The previous import replaced the tree wholesale in a single commit, so there is no shared content history to merge against. Reproducing the rename on the new tree is what actually converges. What survives: .hanzo/workflows, the Dockerfiles and bin/docker* entrypoints, compose files, LLM.md, and insights/migrations. The squashed initial migration is load-bearing — the live database records it by name, so upstream's ~918 individual migrations cannot come back. Nine products are dropped as unreferenced. Twenty-one more that we do not run are kept anyway because the core app still imports them; unpicking those is a refactor and does not belong in a rebase. This base does not build yet, and the reasons are not in the rename: upstream now wants Python 3.13.13 against our 3.12.12 and Django 5.2 against our 4.2, and the dependency manifests still name upstream packages that we republish under our own names. Those are decisions, not renames.
1960 lines
77 KiB
TypeScript
1960 lines
77 KiB
TypeScript
import { Message } from 'node-rdkafka'
|
|
import { v4 } from 'uuid'
|
|
|
|
import {
|
|
DLQ_OUTPUT,
|
|
INGESTION_WARNINGS_OUTPUT,
|
|
IngestionWarningsOutput,
|
|
OVERFLOW_OUTPUT,
|
|
OverflowOutput,
|
|
} from '~/common/outputs'
|
|
import { IngestionOutputs } from '~/common/outputs/ingestion-outputs'
|
|
import { ChunkProcessingStep } from '~/ingestion/framework/base-chunk-pipeline'
|
|
import { newChunkPipelineBuilder } from '~/ingestion/framework/builders'
|
|
import { createBatch, createUnwrapper } from '~/ingestion/framework/helpers'
|
|
import { PipelineConfig } from '~/ingestion/framework/result-handling-pipeline'
|
|
import { PipelineResult, dlq, drop, ok, redirect } from '~/ingestion/framework/results'
|
|
import { ProcessingStep } from '~/ingestion/framework/steps'
|
|
import { createMockIngestionOutputs } from '~/tests/helpers/mock-ingestion-outputs'
|
|
import { ProjectId, Team } from '~/types'
|
|
|
|
const createTestTeam = (overrides: Partial<Team> = {}): Team => ({
|
|
id: 1,
|
|
project_id: 1 as ProjectId,
|
|
organization_id: 'test-org-id',
|
|
uuid: v4(),
|
|
name: 'Test Team',
|
|
anonymize_ips: false,
|
|
api_token: 'test-api-token',
|
|
secret_api_token: null,
|
|
session_recording_opt_in: true,
|
|
person_processing_opt_out: null,
|
|
heatmaps_opt_in: null,
|
|
ingested_event: true,
|
|
person_display_name_properties: null,
|
|
minimal_flag_called_events: false,
|
|
test_account_filters: null,
|
|
cookieless_server_hash_mode: null,
|
|
timezone: 'UTC',
|
|
available_features: [],
|
|
drop_events_older_than_seconds: null,
|
|
extra_settings: null,
|
|
...overrides,
|
|
})
|
|
|
|
// Simple test types - only include fields needed for testing
|
|
type TestEvent = {
|
|
uuid: string
|
|
event: string
|
|
token: string
|
|
chunk_result?: string // Optional field added by chunk processing
|
|
}
|
|
|
|
type TestHeaders = {
|
|
token: string
|
|
}
|
|
|
|
type TestEventWithTeam = {
|
|
message: Message
|
|
event: TestEvent
|
|
team: Team
|
|
headers: TestHeaders
|
|
}
|
|
|
|
/**
|
|
* Integration tests for the pipeline system that approximate the preprocessing pipeline
|
|
* structure used in the ingestion consumer. These tests mirror the key aspects of the
|
|
* ingestion pipeline but are not an exact replica - they focus on testing the core
|
|
* pipeline behavior, ordering guarantees, and result handling patterns.
|
|
*
|
|
* Note: Adding more than 3 preprocessing steps doesn't add meaningful test coverage
|
|
* since the tests already verify the essential pipeline patterns (sync steps, async steps,
|
|
* chunk processing, result handling). The goal is to test pipeline behavior, not to
|
|
* replicate every single step from the actual ingestion consumer.
|
|
*/
|
|
|
|
type AsyncStepConfig<TOutput, R extends string = never> = {
|
|
delay: number
|
|
result: PipelineResult<TOutput, R>
|
|
}
|
|
|
|
const createMockStep = <TInput extends { message: Message }, TOutput, R extends string = never>(
|
|
resultMap: Map<string, AsyncStepConfig<TOutput, R>>
|
|
): jest.MockedFunction<ProcessingStep<TInput, TOutput, R>> => {
|
|
return jest.fn(async (input: TInput) => {
|
|
// Extract event ID from message value
|
|
const eventId = input.message.value?.toString() || 'default'
|
|
|
|
const config = resultMap.get(eventId)
|
|
|
|
if (config) {
|
|
// Simulate async work with configurable delay
|
|
await new Promise((resolve) => setTimeout(resolve, config.delay))
|
|
return config.result
|
|
}
|
|
|
|
// Throw exception if no result found in map
|
|
throw new Error(`No result found for event ID: ${eventId}`)
|
|
})
|
|
}
|
|
|
|
const createMockChunkStep = <TInput, TOutput>(
|
|
resultMap: Map<number, PipelineResult<TOutput>>
|
|
): jest.MockedFunction<ChunkProcessingStep<TInput, TOutput>> => {
|
|
return jest.fn(async (events: TInput[]): Promise<PipelineResult<TOutput>[]> => {
|
|
await new Promise((resolve) => setTimeout(resolve, 1)) // Simulate async work
|
|
return events.map((_, index) => {
|
|
const result = resultMap.get(index)
|
|
if (result) {
|
|
return result
|
|
}
|
|
// Throw exception if no result found in map
|
|
throw new Error(`No result found for event index: ${index}`)
|
|
})
|
|
})
|
|
}
|
|
|
|
describe('Pipeline Integration Tests', () => {
|
|
let mockOutputs: jest.Mocked<
|
|
IngestionOutputs<typeof DLQ_OUTPUT | typeof OVERFLOW_OUTPUT | typeof INGESTION_WARNINGS_OUTPUT>
|
|
>
|
|
let mockPromiseScheduler: any
|
|
let pipelineConfig: PipelineConfig
|
|
|
|
beforeEach(() => {
|
|
mockOutputs = createMockIngestionOutputs<
|
|
typeof DLQ_OUTPUT | typeof OVERFLOW_OUTPUT | typeof INGESTION_WARNINGS_OUTPUT
|
|
>()
|
|
|
|
mockPromiseScheduler = {
|
|
schedule: jest.fn(),
|
|
}
|
|
|
|
pipelineConfig = {
|
|
outputs: mockOutputs,
|
|
promiseScheduler: mockPromiseScheduler,
|
|
}
|
|
})
|
|
|
|
describe('Pipeline Integration Scenarios', () => {
|
|
it('should process events successfully through the pipeline', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('event1'),
|
|
timestamp: Date.now(),
|
|
} as Message,
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 2,
|
|
key: Buffer.from('key2'),
|
|
value: Buffer.from('event2'),
|
|
timestamp: Date.now() + 1,
|
|
} as Message,
|
|
]
|
|
|
|
// Define result maps
|
|
const step1Map = new Map([
|
|
[
|
|
'event1',
|
|
{
|
|
delay: 0,
|
|
result: ok({ message: messages[0], headers: { token: 'test-token' } }),
|
|
},
|
|
],
|
|
[
|
|
'event2',
|
|
{
|
|
delay: 0,
|
|
result: ok({ message: messages[1], headers: { token: 'test-token' } }),
|
|
},
|
|
],
|
|
])
|
|
const step2Map = new Map([
|
|
[
|
|
'event1',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event1', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event2',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event2', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
const step3Map = new Map([
|
|
[
|
|
'event1',
|
|
{
|
|
delay: 10,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event1', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event2',
|
|
{
|
|
delay: 10,
|
|
result: ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event2', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
|
|
// Define chunk step result map
|
|
const chunkStep4Map = new Map([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: {
|
|
uuid: 'event1',
|
|
event: 'test-event-0',
|
|
token: 'test-token',
|
|
chunk_result: 'processed',
|
|
},
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
1,
|
|
ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: {
|
|
uuid: 'event2',
|
|
event: 'test-event-1',
|
|
token: 'test-token',
|
|
chunk_result: 'processed',
|
|
},
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
// Define steps
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }>(step1Map)
|
|
const step2 = createMockStep<
|
|
{ message: Message; headers: TestHeaders },
|
|
{ message: Message; headers: TestHeaders; event: TestEvent }
|
|
>(step2Map)
|
|
const step3 = createMockStep<
|
|
{ message: Message; headers: TestHeaders; event: TestEvent },
|
|
TestEventWithTeam
|
|
>(step3Map)
|
|
|
|
// Define chunk step
|
|
const chunkStep4 = createMockChunkStep<TestEventWithTeam, TestEventWithTeam>(chunkStep4Map)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1).pipe(step2).pipe(step3))
|
|
.gather()
|
|
.pipeChunk(chunkStep4)
|
|
)
|
|
.handleResults(pipelineConfig)
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
const results = await unwrapper.next()
|
|
|
|
expect(results).toHaveLength(2)
|
|
expect((results![0] as TestEventWithTeam).event.event).toBe('test-event-0')
|
|
expect((results![1] as TestEventWithTeam).event.event).toBe('test-event-1')
|
|
|
|
// Verify chunk step added processing metadata
|
|
expect((results![0] as TestEventWithTeam).event.chunk_result).toBe('processed')
|
|
expect((results![1] as TestEventWithTeam).event.chunk_result).toBe('processed')
|
|
|
|
// Verify mock steps were called with correct arguments
|
|
expect(step1).toHaveBeenCalledTimes(2)
|
|
expect(step2).toHaveBeenCalledTimes(2)
|
|
expect(step3).toHaveBeenCalledTimes(2)
|
|
|
|
// Verify chunk step was called with events in correct order
|
|
expect(chunkStep4).toHaveBeenCalledTimes(1)
|
|
expect(chunkStep4).toHaveBeenNthCalledWith(
|
|
1,
|
|
expect.arrayContaining([
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event1' }) }),
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event2' }) }),
|
|
])
|
|
)
|
|
})
|
|
|
|
it('should handle events that are dropped', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('drop-event'),
|
|
timestamp: Date.now(),
|
|
} as Message,
|
|
]
|
|
|
|
// Define result map
|
|
const step1Map = new Map<string, AsyncStepConfig<{ message: Message; headers: TestHeaders }>>([
|
|
['drop-event', { delay: 0, result: drop('Mock drop') }],
|
|
])
|
|
|
|
// Define step
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }>(step1Map)
|
|
|
|
// Define chunk step result map
|
|
const chunkStep2Map = new Map([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'drop-event', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
// Define chunk step
|
|
const chunkStep2 = createMockChunkStep<{ message: Message; headers: TestHeaders }, TestEventWithTeam>(
|
|
chunkStep2Map
|
|
)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1))
|
|
.gather()
|
|
.pipeChunk(chunkStep2)
|
|
)
|
|
.handleResults(pipelineConfig)
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
const results = await unwrapper.next()
|
|
|
|
// Should return empty array since event was dropped
|
|
expect(results).toHaveLength(0)
|
|
|
|
// Verify mock step was called with correct arguments
|
|
expect(step1).toHaveBeenCalledTimes(1)
|
|
expect(step1).toHaveBeenCalledWith({ message: messages[0] })
|
|
})
|
|
|
|
it('should handle events that are redirected', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('redirect-event'),
|
|
timestamp: Date.now(),
|
|
} as Message,
|
|
]
|
|
|
|
// Define result map
|
|
const step1Map = new Map<
|
|
string,
|
|
AsyncStepConfig<{ message: Message; headers: TestHeaders }, OverflowOutput>
|
|
>([['redirect-event', { delay: 0, result: redirect('Mock redirect', OVERFLOW_OUTPUT, true) }]])
|
|
|
|
// Define step
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }, string>(
|
|
step1Map
|
|
)
|
|
|
|
// Define chunk step result map
|
|
const chunkStep2Map = new Map([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'redirect-event', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
// Define chunk step
|
|
const chunkStep2 = createMockChunkStep<{ message: Message; headers: TestHeaders }, TestEventWithTeam>(
|
|
chunkStep2Map
|
|
)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1))
|
|
.gather()
|
|
.pipeChunk(chunkStep2)
|
|
)
|
|
.handleResults(pipelineConfig)
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
const results = await unwrapper.next()
|
|
|
|
// Should return empty array since event was redirected
|
|
expect(results).toHaveLength(0)
|
|
expect(mockOutputs.produce).toHaveBeenCalledWith(
|
|
OVERFLOW_OUTPUT,
|
|
expect.objectContaining({
|
|
value: messages[0].value,
|
|
key: messages[0].key,
|
|
})
|
|
)
|
|
|
|
// Verify mock step was called with correct arguments
|
|
expect(step1).toHaveBeenCalledTimes(1)
|
|
expect(step1).toHaveBeenCalledWith({ message: messages[0] })
|
|
})
|
|
|
|
it('should handle events that go to DLQ', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('dlq-event'),
|
|
timestamp: Date.now(),
|
|
} as Message,
|
|
]
|
|
|
|
// Define result map
|
|
const step1Map = new Map<
|
|
string,
|
|
AsyncStepConfig<{ message: Message; headers: TestHeaders; event: TestEvent }>
|
|
>([['dlq-event', { delay: 0, result: dlq('Mock DLQ', new Error('Mock error')) }]])
|
|
|
|
// Define step
|
|
const step1 = createMockStep<
|
|
{ message: Message },
|
|
{ message: Message; headers: TestHeaders; event: TestEvent }
|
|
>(step1Map)
|
|
|
|
// Define chunk step result map
|
|
const chunkStep2Map = new Map([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'dlq-event', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
// Define chunk step
|
|
const chunkStep2 = createMockChunkStep<
|
|
{ message: Message; headers: TestHeaders; event: TestEvent },
|
|
TestEventWithTeam
|
|
>(chunkStep2Map)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1))
|
|
.gather()
|
|
.pipeChunk(chunkStep2)
|
|
)
|
|
.handleResults(pipelineConfig)
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
const results = await unwrapper.next()
|
|
|
|
// Should return empty array since event went to DLQ
|
|
expect(results).toHaveLength(0)
|
|
expect(mockOutputs.produce).toHaveBeenCalledWith(
|
|
DLQ_OUTPUT,
|
|
expect.objectContaining({
|
|
value: messages[0].value,
|
|
key: messages[0].key,
|
|
})
|
|
)
|
|
|
|
// Verify mock step was called with correct arguments
|
|
expect(step1).toHaveBeenCalledTimes(1)
|
|
expect(step1).toHaveBeenCalledWith({ message: messages[0] })
|
|
})
|
|
|
|
it('should process events concurrently while preserving final ordering', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('event-1'),
|
|
timestamp: Date.now(),
|
|
} as Message,
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 2,
|
|
key: Buffer.from('key2'),
|
|
value: Buffer.from('event-2'),
|
|
timestamp: Date.now() + 1,
|
|
} as Message,
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 3,
|
|
key: Buffer.from('key3'),
|
|
value: Buffer.from('event-3'),
|
|
timestamp: Date.now() + 2,
|
|
} as Message,
|
|
]
|
|
|
|
// Define result maps
|
|
const step1Map = new Map([
|
|
['event-1', { delay: 0, result: ok({ message: messages[0], headers: { token: 'test-token' } }) }],
|
|
['event-2', { delay: 0, result: ok({ message: messages[1], headers: { token: 'test-token' } }) }],
|
|
['event-3', { delay: 0, result: ok({ message: messages[2], headers: { token: 'test-token' } }) }],
|
|
])
|
|
const step2Map = new Map([
|
|
[
|
|
'event-1',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-1', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-2',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-2', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-3',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-3', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
const step3Map = new Map([
|
|
[
|
|
'event-1',
|
|
{
|
|
delay: 5,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-1', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-2',
|
|
{
|
|
delay: 15,
|
|
result: ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-2', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-3',
|
|
{
|
|
delay: 10,
|
|
result: ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-3', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
|
|
// Define chunk step result map
|
|
const chunkStep4Map = new Map([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-1', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
1,
|
|
ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-2', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
2,
|
|
ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-3', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
// Define steps
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }>(step1Map)
|
|
const step2 = createMockStep<
|
|
{ message: Message; headers: TestHeaders },
|
|
{ message: Message; headers: TestHeaders; event: TestEvent }
|
|
>(step2Map)
|
|
const step3 = createMockStep<
|
|
{ message: Message; headers: TestHeaders; event: TestEvent },
|
|
TestEventWithTeam
|
|
>(step3Map)
|
|
|
|
// Define chunk step
|
|
const chunkStep4 = createMockChunkStep<TestEventWithTeam, TestEventWithTeam>(chunkStep4Map)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1).pipe(step2).pipe(step3))
|
|
.gather()
|
|
.pipeChunk(chunkStep4)
|
|
)
|
|
.handleResults(pipelineConfig)
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
const results = await unwrapper.next()
|
|
|
|
expect(results).toHaveLength(3)
|
|
// Verify final ordering is preserved despite concurrent preprocessing
|
|
expect((results![0] as TestEventWithTeam).event.uuid).toBe('event-1')
|
|
expect((results![1] as TestEventWithTeam).event.uuid).toBe('event-2')
|
|
expect((results![2] as TestEventWithTeam).event.uuid).toBe('event-3')
|
|
|
|
// Verify chunk step added processing metadata to all events
|
|
expect((results![0] as TestEventWithTeam).event.chunk_result).toBe('processed')
|
|
expect((results![1] as TestEventWithTeam).event.chunk_result).toBe('processed')
|
|
expect((results![2] as TestEventWithTeam).event.chunk_result).toBe('processed')
|
|
|
|
// Verify mock steps were called with correct arguments
|
|
expect(step1).toHaveBeenCalledTimes(3)
|
|
expect(step2).toHaveBeenCalledTimes(3)
|
|
expect(step3).toHaveBeenCalledTimes(3)
|
|
|
|
// Verify chunk step was called with events in correct order
|
|
expect(chunkStep4).toHaveBeenCalledTimes(1)
|
|
expect(chunkStep4).toHaveBeenNthCalledWith(
|
|
1,
|
|
expect.arrayContaining([
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event-1' }) }),
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event-2' }) }),
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event-3' }) }),
|
|
])
|
|
)
|
|
})
|
|
|
|
it('should handle mixed success and failure scenarios in chunk', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('event1'),
|
|
timestamp: Date.now(),
|
|
} as Message,
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 2,
|
|
key: Buffer.from('key2'),
|
|
value: Buffer.from('drop-event'),
|
|
timestamp: Date.now() + 1,
|
|
} as Message,
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 3,
|
|
key: Buffer.from('key3'),
|
|
value: Buffer.from('event3'),
|
|
timestamp: Date.now() + 2,
|
|
} as Message,
|
|
]
|
|
|
|
// Define result maps
|
|
const step1Map = new Map<string, AsyncStepConfig<{ message: Message; headers: TestHeaders }>>([
|
|
['event1', { delay: 0, result: ok({ message: messages[0], headers: { token: 'test-token' } }) }],
|
|
['drop-event', { delay: 0, result: drop('Mock drop') }],
|
|
['event3', { delay: 0, result: ok({ message: messages[2], headers: { token: 'test-token' } }) }],
|
|
])
|
|
const step2Map = new Map([
|
|
[
|
|
'event1',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event1', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event3',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event3', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
const step3Map = new Map([
|
|
[
|
|
'event1',
|
|
{
|
|
delay: 8,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event1', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event3',
|
|
{
|
|
delay: 6,
|
|
result: ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event3', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
|
|
// Define chunk step result map
|
|
const chunkStep4Map = new Map([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event1', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
1,
|
|
ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event3', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
// Define steps
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }>(step1Map)
|
|
const step2 = createMockStep<
|
|
{ message: Message; headers: TestHeaders },
|
|
{ message: Message; headers: TestHeaders; event: TestEvent }
|
|
>(step2Map)
|
|
const step3 = createMockStep<
|
|
{ message: Message; headers: TestHeaders; event: TestEvent },
|
|
TestEventWithTeam
|
|
>(step3Map)
|
|
|
|
// Define chunk step
|
|
const chunkStep4 = createMockChunkStep<TestEventWithTeam, TestEventWithTeam>(chunkStep4Map)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1).pipe(step2).pipe(step3))
|
|
.gather()
|
|
.pipeChunk(chunkStep4)
|
|
)
|
|
.handleResults(pipelineConfig)
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
const results = await unwrapper.next()
|
|
|
|
// Should only return the valid events
|
|
expect(results).toHaveLength(2)
|
|
expect((results![0] as TestEventWithTeam).event.uuid).toBe('event1')
|
|
expect((results![1] as TestEventWithTeam).event.uuid).toBe('event3')
|
|
|
|
// Verify chunk step added processing metadata
|
|
expect((results![0] as TestEventWithTeam).event.chunk_result).toBe('processed')
|
|
expect((results![1] as TestEventWithTeam).event.chunk_result).toBe('processed')
|
|
|
|
// Verify mock steps were called with correct arguments
|
|
expect(step1).toHaveBeenCalledTimes(3)
|
|
expect(step2).toHaveBeenCalledTimes(2) // Only valid events reach validation
|
|
expect(step3).toHaveBeenCalledTimes(2) // Only valid events reach team resolution
|
|
expect(chunkStep4).toHaveBeenCalledTimes(1) // Chunk step called once with 2 events
|
|
|
|
// Verify chunk step was called with remaining events in correct order (dropped event excluded)
|
|
expect(chunkStep4).toHaveBeenNthCalledWith(
|
|
1,
|
|
expect.arrayContaining([
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event1' }) }),
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event3' }) }),
|
|
])
|
|
)
|
|
})
|
|
})
|
|
|
|
describe('Error Handling', () => {
|
|
it('should handle exceptions in preprocessing steps', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('event1'),
|
|
timestamp: 1234567890,
|
|
size: 6,
|
|
},
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 2,
|
|
key: Buffer.from('key2'),
|
|
value: Buffer.from('event2'),
|
|
timestamp: 1234567891,
|
|
size: 6,
|
|
},
|
|
]
|
|
|
|
// Define result maps
|
|
const step1Map = new Map([
|
|
['event1', { delay: 0, result: ok({ message: messages[0], headers: { token: 'test-token' } }) }],
|
|
['event2', { delay: 0, result: ok({ message: messages[1], headers: { token: 'test-token' } }) }],
|
|
])
|
|
|
|
// Define chunk step result map
|
|
const chunkStep4Map = new Map([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: {
|
|
uuid: 'event1',
|
|
event: 'test-event-0',
|
|
token: 'test-token',
|
|
chunk_result: 'processed',
|
|
},
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
1,
|
|
ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: {
|
|
uuid: 'event2',
|
|
event: 'test-event-1',
|
|
token: 'test-token',
|
|
chunk_result: 'processed',
|
|
},
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
// Define steps
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }>(step1Map)
|
|
|
|
// Create a step that throws an exception for one event
|
|
const step2 = jest.fn(
|
|
async (input: {
|
|
message: Message
|
|
headers: TestHeaders
|
|
}): Promise<PipelineResult<{ message: Message; headers: TestHeaders; event: TestEvent }>> => {
|
|
await new Promise((resolve) => setTimeout(resolve, 1)) // Simulate async work
|
|
const eventId = input.message.value?.toString() || 'unknown'
|
|
if (eventId === 'event1') {
|
|
throw new Error('Mock preprocessing exception')
|
|
}
|
|
return ok({
|
|
message: input.message,
|
|
headers: input.headers,
|
|
event: { uuid: eventId, event: 'test-event', token: 'test-token' },
|
|
})
|
|
}
|
|
)
|
|
|
|
const step3Map = new Map<string, AsyncStepConfig<TestEventWithTeam>>([
|
|
[
|
|
'event2',
|
|
{
|
|
delay: 10,
|
|
result: ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event2', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
const step3 = createMockStep<
|
|
{ message: Message; headers: TestHeaders; event: TestEvent },
|
|
TestEventWithTeam
|
|
>(step3Map)
|
|
|
|
const chunkStep4 = createMockChunkStep<TestEventWithTeam, TestEventWithTeam>(chunkStep4Map)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1).pipe(step2).pipe(step3))
|
|
.gather()
|
|
.pipeChunk(chunkStep4)
|
|
)
|
|
.handleResults({
|
|
outputs: mockOutputs,
|
|
promiseScheduler: mockPromiseScheduler,
|
|
})
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
// Expect the pipeline to throw an exception
|
|
await expect(unwrapper.next()).rejects.toThrow('Mock preprocessing exception')
|
|
|
|
// Verify preprocessing steps were called
|
|
expect(step1).toHaveBeenCalledTimes(2)
|
|
expect(step2).toHaveBeenCalledTimes(2)
|
|
// Note: step3 call count is not deterministic due to concurrent processing and exception handling
|
|
|
|
// Verify chunk step was not called due to exception
|
|
expect(chunkStep4).toHaveBeenCalledTimes(0)
|
|
})
|
|
|
|
it('should handle exceptions in chunk steps', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('event1'),
|
|
timestamp: 1234567890,
|
|
size: 6,
|
|
},
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 2,
|
|
key: Buffer.from('key2'),
|
|
value: Buffer.from('event2'),
|
|
timestamp: 1234567891,
|
|
size: 6,
|
|
},
|
|
]
|
|
|
|
// Define result maps
|
|
const step1Map = new Map([
|
|
[
|
|
'event1',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event1', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event2',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event2', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
|
|
// Define chunk step result maps
|
|
const chunkStep1Map = new Map([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: {
|
|
uuid: 'event1',
|
|
event: 'test-event-0',
|
|
token: 'test-token',
|
|
chunk_result: 'processed',
|
|
},
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
1,
|
|
ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: {
|
|
uuid: 'event2',
|
|
event: 'test-event-1',
|
|
token: 'test-token',
|
|
chunk_result: 'processed',
|
|
},
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
const chunkStep3Map = new Map<number, PipelineResult<TestEventWithTeam>>([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: {
|
|
uuid: 'event1',
|
|
event: 'test-event-0',
|
|
token: 'test-token',
|
|
chunk_result: 'processed',
|
|
},
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
1,
|
|
ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: {
|
|
uuid: 'event2',
|
|
event: 'test-event-1',
|
|
token: 'test-token',
|
|
chunk_result: 'processed',
|
|
},
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
// Define steps
|
|
const step1 = createMockStep<{ message: Message }, TestEventWithTeam>(step1Map)
|
|
|
|
const chunkStep1 = createMockChunkStep<TestEventWithTeam, TestEventWithTeam>(chunkStep1Map)
|
|
|
|
// Create a chunk step that throws an exception
|
|
const chunkStep2 = jest.fn(
|
|
async (_events: TestEventWithTeam[]): Promise<PipelineResult<TestEventWithTeam>[]> => {
|
|
await new Promise((resolve) => setTimeout(resolve, 1)) // Simulate async work
|
|
throw new Error('Mock chunk exception')
|
|
}
|
|
)
|
|
|
|
const chunkStep3 = createMockChunkStep<TestEventWithTeam, TestEventWithTeam>(chunkStep3Map)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1))
|
|
.gather()
|
|
.pipeChunk(chunkStep1)
|
|
.pipeChunk(chunkStep2)
|
|
.pipeChunk(chunkStep3)
|
|
)
|
|
.handleResults({
|
|
outputs: mockOutputs,
|
|
promiseScheduler: mockPromiseScheduler,
|
|
})
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
// Expect the pipeline to throw an exception
|
|
await expect(unwrapper.next()).rejects.toThrow('Mock chunk exception')
|
|
|
|
// Verify preprocessing step was called
|
|
expect(step1).toHaveBeenCalledTimes(2)
|
|
|
|
// Verify chunk steps were called up to the exception
|
|
expect(chunkStep1).toHaveBeenCalledTimes(1)
|
|
expect(chunkStep2).toHaveBeenCalledTimes(1)
|
|
|
|
// Verify third chunk step was not called due to exception
|
|
expect(chunkStep3).toHaveBeenCalledTimes(0)
|
|
})
|
|
|
|
it('should handle mixed results and exceptions without producing to Kafka', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('drop-event'),
|
|
timestamp: 1234567890,
|
|
size: 10,
|
|
},
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 2,
|
|
key: Buffer.from('key2'),
|
|
value: Buffer.from('dlq-event'),
|
|
timestamp: 1234567891,
|
|
size: 8,
|
|
},
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 3,
|
|
key: Buffer.from('key3'),
|
|
value: Buffer.from('redirect-event'),
|
|
timestamp: 1234567892,
|
|
size: 13,
|
|
},
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 4,
|
|
key: Buffer.from('key4'),
|
|
value: Buffer.from('ok-event'),
|
|
timestamp: 1234567893,
|
|
size: 7,
|
|
},
|
|
]
|
|
|
|
// Define result maps
|
|
const step1Map = new Map<
|
|
string,
|
|
AsyncStepConfig<{ message: Message; headers: TestHeaders }, OverflowOutput>
|
|
>([
|
|
['drop-event', { delay: 0, result: drop('Mock drop') }],
|
|
['dlq-event', { delay: 0, result: dlq('Mock DLQ') }],
|
|
['redirect-event', { delay: 0, result: redirect('Mock redirect', OVERFLOW_OUTPUT, true) }],
|
|
['ok-event', { delay: 0, result: ok({ message: messages[3], headers: { token: 'test-token' } }) }],
|
|
])
|
|
|
|
// Define steps
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }, string>(
|
|
step1Map
|
|
)
|
|
|
|
// Create a step that throws an exception for the remaining event
|
|
const step2 = jest.fn(
|
|
async (_input: {
|
|
message: Message
|
|
headers: TestHeaders
|
|
}): Promise<PipelineResult<{ message: Message; headers: TestHeaders; event: TestEvent }>> => {
|
|
await new Promise((resolve) => setTimeout(resolve, 1)) // Simulate async work
|
|
throw new Error('Mock step2 exception')
|
|
}
|
|
)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) => builder.concurrently((b) => b.pipe(step1).pipe(step2)).gather())
|
|
.handleResults({
|
|
outputs: mockOutputs,
|
|
promiseScheduler: mockPromiseScheduler,
|
|
})
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
// Expect the pipeline to throw an exception
|
|
await expect(unwrapper.next()).rejects.toThrow('Mock step2 exception')
|
|
|
|
// Verify preprocessing steps were called
|
|
expect(step1).toHaveBeenCalledTimes(4)
|
|
expect(step2).toHaveBeenCalledTimes(1) // Only ok-event reaches step2
|
|
|
|
// Verify no messages were produced due to exception
|
|
expect(mockOutputs.produce).toHaveBeenCalledTimes(0)
|
|
})
|
|
})
|
|
|
|
describe('Concurrent Processing', () => {
|
|
it('should preserve ordering with concurrently and gather using deterministic mocks', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('event1'),
|
|
timestamp: 1234567890,
|
|
size: 6,
|
|
},
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 2,
|
|
key: Buffer.from('key2'),
|
|
value: Buffer.from('event2'),
|
|
timestamp: 1234567891,
|
|
size: 6,
|
|
},
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 3,
|
|
key: Buffer.from('key3'),
|
|
value: Buffer.from('event3'),
|
|
timestamp: 1234567892,
|
|
size: 6,
|
|
},
|
|
]
|
|
|
|
// Create promises with stored resolve callbacks for manual control
|
|
let resolve1: () => void
|
|
let resolve2: () => void
|
|
let resolve3: () => void
|
|
|
|
const promise1 = new Promise<void>((resolve) => {
|
|
resolve1 = resolve
|
|
})
|
|
const promise2 = new Promise<void>((resolve) => {
|
|
resolve2 = resolve
|
|
})
|
|
const promise3 = new Promise<void>((resolve) => {
|
|
resolve3 = resolve
|
|
})
|
|
|
|
// Define result maps
|
|
const step1Map = new Map<string, AsyncStepConfig<{ message: Message; headers: TestHeaders }>>([
|
|
['event1', { delay: 0, result: ok({ message: messages[0], headers: { token: 'test-token' } }) }],
|
|
['event2', { delay: 0, result: ok({ message: messages[1], headers: { token: 'test-token' } }) }],
|
|
['event3', { delay: 0, result: ok({ message: messages[2], headers: { token: 'test-token' } }) }],
|
|
])
|
|
|
|
// Create deterministic async step that waits for manual resolution
|
|
let callCount = 0
|
|
const step2 = jest.fn(
|
|
async (input: {
|
|
message: Message
|
|
headers: TestHeaders
|
|
}): Promise<PipelineResult<TestEventWithTeam>> => {
|
|
const eventId = input.message.value?.toString() || 'unknown'
|
|
const currentCall = callCount++
|
|
|
|
// Wait for the appropriate promise to be manually resolved
|
|
if (currentCall === 0) {
|
|
await promise1
|
|
} else if (currentCall === 1) {
|
|
await promise2
|
|
} else if (currentCall === 2) {
|
|
await promise3
|
|
}
|
|
|
|
return ok({
|
|
message: input.message,
|
|
headers: input.headers,
|
|
event: { uuid: eventId, event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
})
|
|
}
|
|
)
|
|
|
|
// Define steps
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }>(step1Map)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) => builder.concurrently((b) => b.pipe(step1).pipe(step2)).gather())
|
|
.handleResults({
|
|
outputs: mockOutputs,
|
|
promiseScheduler: mockPromiseScheduler,
|
|
})
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
// Start the pipeline processing
|
|
const resultsPromise = unwrapper.next()
|
|
|
|
// Resolve promises in specific order to test ordering preservation
|
|
// 1. Resolve second promise first
|
|
resolve2!()
|
|
await new Promise((resolve) => setImmediate(resolve)) // Wait for next tick
|
|
|
|
// Check that results promise is not resolved yet (waiting for other promises)
|
|
const isResolvedAfter2 = await Promise.race([
|
|
resultsPromise.then(() => true),
|
|
new Promise((resolve) => setImmediate(() => resolve(false))),
|
|
])
|
|
expect(isResolvedAfter2).toBe(false)
|
|
|
|
// 2. Resolve first promise
|
|
resolve1!()
|
|
await new Promise((resolve) => setImmediate(resolve)) // Wait for next tick
|
|
|
|
// Check that results promise is still not resolved (waiting for third promise)
|
|
const isResolvedAfter1 = await Promise.race([
|
|
resultsPromise.then(() => true),
|
|
new Promise((resolve) => setImmediate(() => resolve(false))),
|
|
])
|
|
expect(isResolvedAfter1).toBe(false)
|
|
|
|
// 3. Resolve third promise last
|
|
resolve3!()
|
|
await new Promise((resolve) => setImmediate(resolve)) // Wait for next tick
|
|
|
|
const results = await resultsPromise
|
|
|
|
expect(results).toHaveLength(3)
|
|
|
|
// Verify ordering is preserved despite concurrent processing and manual resolution order
|
|
expect((results![0] as TestEventWithTeam).event.uuid).toBe('event1')
|
|
expect((results![1] as TestEventWithTeam).event.uuid).toBe('event2')
|
|
expect((results![2] as TestEventWithTeam).event.uuid).toBe('event3')
|
|
|
|
// Verify mock steps were called with correct arguments
|
|
expect(step1).toHaveBeenCalledTimes(3)
|
|
expect(step2).toHaveBeenCalledTimes(3)
|
|
})
|
|
|
|
it('should process multiple events concurrently while maintaining ordering', async () => {
|
|
const messages: Message[] = Array.from(
|
|
{ length: 5 },
|
|
(_, i) =>
|
|
({
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: i + 1,
|
|
key: Buffer.from(`key${i}`),
|
|
value: Buffer.from(`event-${i}`),
|
|
timestamp: Date.now() + i,
|
|
}) as Message
|
|
)
|
|
|
|
// Define result maps
|
|
const step1Map = new Map([
|
|
['event-0', { delay: 0, result: ok({ message: messages[0], headers: { token: 'test-token' } }) }],
|
|
['event-1', { delay: 0, result: ok({ message: messages[1], headers: { token: 'test-token' } }) }],
|
|
['event-2', { delay: 0, result: ok({ message: messages[2], headers: { token: 'test-token' } }) }],
|
|
['event-3', { delay: 0, result: ok({ message: messages[3], headers: { token: 'test-token' } }) }],
|
|
['event-4', { delay: 0, result: ok({ message: messages[4], headers: { token: 'test-token' } }) }],
|
|
])
|
|
const step2Map = new Map([
|
|
[
|
|
'event-0',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-0', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-1',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-1', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-2',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-2', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-3',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[3],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-3', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-4',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[4],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-4', event: 'test-event', token: 'test-token' },
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
const step3Map = new Map([
|
|
[
|
|
'event-0',
|
|
{
|
|
delay: 20,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-0', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-1',
|
|
{
|
|
delay: 15,
|
|
result: ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-1', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-2',
|
|
{
|
|
delay: 25,
|
|
result: ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-2', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-3',
|
|
{
|
|
delay: 10,
|
|
result: ok({
|
|
message: messages[3],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-3', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event-4',
|
|
{
|
|
delay: 30,
|
|
result: ok({
|
|
message: messages[4],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-4', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
|
|
// Define chunk step result map
|
|
const chunkStep4Map = new Map([
|
|
[
|
|
0,
|
|
ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-0', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
1,
|
|
ok({
|
|
message: messages[1],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-1', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
2,
|
|
ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-2', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
3,
|
|
ok({
|
|
message: messages[3],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-3', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
[
|
|
4,
|
|
ok({
|
|
message: messages[4],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event-4', event: 'test-event', token: 'test-token', chunk_result: 'processed' },
|
|
team: createTestTeam(),
|
|
}),
|
|
],
|
|
])
|
|
|
|
// Define steps
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }>(step1Map)
|
|
const step2 = createMockStep<
|
|
{ message: Message; headers: TestHeaders },
|
|
{ message: Message; headers: TestHeaders; event: TestEvent }
|
|
>(step2Map)
|
|
const step3 = createMockStep<
|
|
{ message: Message; headers: TestHeaders; event: TestEvent },
|
|
TestEventWithTeam
|
|
>(step3Map)
|
|
|
|
// Define chunk step
|
|
const chunkStep4 = createMockChunkStep<TestEventWithTeam, TestEventWithTeam>(chunkStep4Map)
|
|
|
|
// Create pipeline
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1).pipe(step2).pipe(step3))
|
|
.gather()
|
|
.pipeChunk(chunkStep4)
|
|
)
|
|
.handleResults(pipelineConfig)
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
const results = await unwrapper.next()
|
|
|
|
expect(results).toHaveLength(5)
|
|
// Verify final ordering is preserved despite concurrent preprocessing
|
|
for (let i = 0; i < 5; i++) {
|
|
expect((results![i] as TestEventWithTeam).event.uuid).toBe(`event-${i}`)
|
|
}
|
|
|
|
// Verify mock steps were called with correct arguments
|
|
expect(step1).toHaveBeenCalledTimes(5)
|
|
expect(step2).toHaveBeenCalledTimes(5)
|
|
expect(step3).toHaveBeenCalledTimes(5)
|
|
|
|
// Verify chunk step was called with events in correct order
|
|
expect(chunkStep4).toHaveBeenCalledTimes(1)
|
|
expect(chunkStep4).toHaveBeenNthCalledWith(
|
|
1,
|
|
expect.arrayContaining([
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event-0' }) }),
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event-1' }) }),
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event-2' }) }),
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event-3' }) }),
|
|
expect.objectContaining({ event: expect.objectContaining({ uuid: 'event-4' }) }),
|
|
])
|
|
)
|
|
})
|
|
})
|
|
|
|
describe('Builder API with filterMap and teamAware', () => {
|
|
it('should filter OK results, map context to include team, and process team-aware steps', async () => {
|
|
const messages: Message[] = [
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 1,
|
|
key: Buffer.from('key1'),
|
|
value: Buffer.from('event1'),
|
|
timestamp: Date.now(),
|
|
size: 6,
|
|
},
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 2,
|
|
key: Buffer.from('key2'),
|
|
value: Buffer.from('drop-event'),
|
|
timestamp: Date.now() + 1,
|
|
size: 10,
|
|
},
|
|
{
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: 3,
|
|
key: Buffer.from('key3'),
|
|
value: Buffer.from('event3'),
|
|
timestamp: Date.now() + 2,
|
|
size: 6,
|
|
},
|
|
]
|
|
|
|
// Define result maps
|
|
const step1Map = new Map<string, AsyncStepConfig<{ message: Message; headers: TestHeaders }>>([
|
|
['event1', { delay: 0, result: ok({ message: messages[0], headers: { token: 'test-token' } }) }],
|
|
['drop-event', { delay: 0, result: drop('Mock drop') }],
|
|
['event3', { delay: 0, result: ok({ message: messages[2], headers: { token: 'test-token' } }) }],
|
|
])
|
|
const step2Map = new Map([
|
|
[
|
|
'event1',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event1', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam({ id: 1, name: 'Test Team 1' }),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event3',
|
|
{
|
|
delay: 0,
|
|
result: ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event3', event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam({ id: 2, name: 'Test Team 2' }),
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
|
|
const step3Map = new Map([
|
|
[
|
|
'event1',
|
|
{
|
|
delay: 5,
|
|
result: ok({
|
|
message: messages[0],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event1', event: 'validated-event', token: 'test-token' },
|
|
team: createTestTeam({ id: 1, name: 'Test Team 1' }),
|
|
}),
|
|
},
|
|
],
|
|
[
|
|
'event3',
|
|
{
|
|
delay: 3,
|
|
result: ok({
|
|
message: messages[2],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: 'event3', event: 'validated-event', token: 'test-token' },
|
|
team: createTestTeam({ id: 2, name: 'Test Team 2' }),
|
|
}),
|
|
},
|
|
],
|
|
])
|
|
|
|
// Define steps
|
|
const step1 = createMockStep<{ message: Message }, { message: Message; headers: TestHeaders }>(step1Map)
|
|
const step2 = createMockStep<{ message: Message; headers: TestHeaders }, TestEventWithTeam>(step2Map)
|
|
const step3 = createMockStep<TestEventWithTeam, TestEventWithTeam>(step3Map)
|
|
// Chunk step adds chunk_result to each event - uses input content instead of index
|
|
// because filterMap feeds events in separate chunks (not gathered)
|
|
const chunkStep: jest.MockedFunction<ChunkProcessingStep<TestEventWithTeam, TestEventWithTeam>> = jest.fn(
|
|
(events) =>
|
|
Promise.resolve(
|
|
events.map((event) => ok({ ...event, event: { ...event.event, chunk_result: 'processed' } }))
|
|
)
|
|
)
|
|
|
|
// Create pipeline using builder API
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1).pipe(step2))
|
|
.filterMap(
|
|
(element) => ({
|
|
result: element.result,
|
|
context: {
|
|
...element.context,
|
|
team: element.result.value.team,
|
|
},
|
|
}),
|
|
(b) =>
|
|
b
|
|
.teamAware((b) =>
|
|
b
|
|
.concurrently((c) => c.pipe(step3))
|
|
.gather()
|
|
.pipeChunk(chunkStep)
|
|
)
|
|
.handleIngestionWarnings(createMockIngestionOutputs<IngestionWarningsOutput>())
|
|
)
|
|
)
|
|
.handleResults(pipelineConfig)
|
|
.handleSideEffects(mockPromiseScheduler, { await: false })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
const allResults: any[] = []
|
|
let results = await unwrapper.next()
|
|
while (results !== null) {
|
|
allResults.push(...results)
|
|
results = await unwrapper.next()
|
|
}
|
|
|
|
// Verify steps were called correctly
|
|
expect(step1).toHaveBeenCalledTimes(3) // All 3 events
|
|
expect(step2).toHaveBeenCalledTimes(2) // Only non-dropped events
|
|
expect(step3).toHaveBeenCalledTimes(2) // Only team-aware events
|
|
expect(chunkStep).toHaveBeenCalledTimes(2) // Chunk processed per concurrent result chunk
|
|
|
|
// Should only return the OK events (drop-event filtered out)
|
|
expect(allResults).toHaveLength(2)
|
|
expect(allResults![0].event.uuid).toBe('event1')
|
|
expect(allResults![1].event.uuid).toBe('event3')
|
|
|
|
// Verify chunk processing metadata
|
|
expect(allResults![0].event.chunk_result).toBe('processed')
|
|
expect(allResults![1].event.chunk_result).toBe('processed')
|
|
|
|
// Verify teams are correctly mapped
|
|
expect(allResults![0].team.id).toBe(1)
|
|
expect(allResults![1].team.id).toBe(2)
|
|
})
|
|
})
|
|
|
|
describe('Performance and Concurrency', () => {
|
|
it('should process 100 events with random delays efficiently', async () => {
|
|
const messages: Message[] = Array.from(
|
|
{ length: 100 },
|
|
(_, i) =>
|
|
({
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
offset: i + 1,
|
|
key: Buffer.from(`key${i}`),
|
|
value: Buffer.from(`event-${i}`),
|
|
timestamp: Date.now() + i,
|
|
}) as Message
|
|
)
|
|
|
|
// Create result map with random delays between 50-150ms
|
|
const step1Map = new Map<string, AsyncStepConfig<TestEventWithTeam>>()
|
|
for (let i = 0; i < 100; i++) {
|
|
const randomDelay = Math.floor(Math.random() * 100) + 50 // 50-150ms
|
|
step1Map.set(`event-${i}`, {
|
|
delay: randomDelay,
|
|
result: ok({
|
|
message: messages[i],
|
|
headers: { token: 'test-token' },
|
|
event: { uuid: `event-${i}`, event: 'test-event', token: 'test-token' },
|
|
team: createTestTeam(),
|
|
}),
|
|
})
|
|
}
|
|
|
|
// Define chunk step result map
|
|
const chunkStep2Map = new Map<number, PipelineResult<TestEventWithTeam>>()
|
|
for (let i = 0; i < 100; i++) {
|
|
chunkStep2Map.set(
|
|
i,
|
|
ok({
|
|
message: messages[i],
|
|
headers: { token: 'test-token' },
|
|
event: {
|
|
uuid: `event-${i}`,
|
|
event: 'test-event',
|
|
token: 'test-token',
|
|
chunk_result: 'processed',
|
|
},
|
|
team: createTestTeam(),
|
|
})
|
|
)
|
|
}
|
|
|
|
// Define step with random delays
|
|
const step1 = createMockStep<{ message: Message }, TestEventWithTeam>(step1Map)
|
|
|
|
// Define chunk step
|
|
const chunkStep2 = createMockChunkStep<TestEventWithTeam, TestEventWithTeam>(chunkStep2Map)
|
|
|
|
// Create pipeline with single async step
|
|
|
|
const pipeline = newChunkPipelineBuilder<{ message: Message }, { message: Message }>()
|
|
.messageAware((builder) =>
|
|
builder
|
|
.concurrently((b) => b.pipe(step1))
|
|
.gather()
|
|
.pipeChunk(chunkStep2)
|
|
)
|
|
.handleResults(pipelineConfig)
|
|
.handleSideEffects(mockPromiseScheduler, { await: true })
|
|
.build()
|
|
|
|
const unwrapper = createUnwrapper(pipeline)
|
|
|
|
const batch = createBatch(messages.map((message) => ({ message })))
|
|
unwrapper.feed(batch)
|
|
|
|
const startTime = Date.now()
|
|
const results = await unwrapper.next()
|
|
const endTime = Date.now()
|
|
|
|
// Verify all events were processed
|
|
expect(results).toHaveLength(100)
|
|
|
|
// Verify ordering is preserved
|
|
for (let i = 0; i < 100; i++) {
|
|
expect((results![i] as TestEventWithTeam).event.uuid).toBe(`event-${i}`)
|
|
}
|
|
|
|
// Assert pipeline completes within reasonable time (50-300ms with safety margin)
|
|
const totalTime = endTime - startTime
|
|
expect(totalTime).toBeGreaterThanOrEqual(50) // At least 50ms due to minimum delay
|
|
expect(totalTime).toBeLessThanOrEqual(300) // Should complete within 300ms due to concurrency
|
|
|
|
// Verify mock step was called for all events
|
|
expect(step1).toHaveBeenCalledTimes(100)
|
|
})
|
|
})
|
|
})
|