floci-ui/packages/api/src/adapter-aws/AwsPipesAdapter.test.ts

import {describe, expect, test} from 'bun:test'
import {
    CreatePipeCommand,
    DeletePipeCommand,
    DescribePipeCommand,
    ListPipesCommand,
    type PipesClient,
    StartPipeCommand,
    StopPipeCommand,
} from '@aws-sdk/client-pipes'
import {AwsPipesAdapter, describeArn} from './AwsPipesAdapter'
import {ValidationError} from '../cloud-spi/errors'

type SendResult = Record<string, unknown>

function stubPipes(handler: (command: object) => SendResult | Promise<SendResult>) {
    const sent: object[] = []
    const client = {
        async send(command: object) {
            sent.push(command)
            return handler(command)
        },
    } as unknown as PipesClient
    return {client, sent, adapter: new AwsPipesAdapter(client)}
}

const SRC = 'arn:aws:sqs:us-east-1:000000000000:pipe-src'
const DST = 'arn:aws:lambda:us-east-1:000000000000:function:worker'
const ROLE = 'arn:aws:iam::000000000000:role/pipe-role'
const described = {
    Name: 'orders-pipe', Arn: 'arn:aws:pipes:us-east-1:000000000000:pipe/orders-pipe',
    DesiredState: 'RUNNING', CurrentState: 'RUNNING', Source: SRC, Target: DST, RoleArn: ROLE,
    SourceParameters: {FilterCriteria: {Filters: [{Pattern: '{"body":{"kind":["order"]}}'}]}, SqsQueueParameters: {BatchSize: 5}},
    TargetParameters: {InputTemplate: '{"id": <$.messageId>}'},
    CreationTime: new Date('2026-10-25T16:42:45Z'),
}

describe('AwsPipesAdapter', () => {
    test('lists pipes across pages with readable source and target names', async () => {
        const {adapter} = stubPipes((command) => {
            if (!(command instanceof ListPipesCommand)) return {}
            return command.input.NextToken
                ? {Pipes: [{Name: 'clicks', CurrentState: 'STOPPED', Source: 'arn:aws:kinesis:us-east-1:000000000000:stream/clicks', Target: 'arn:aws:events:us-east-1:000000000000:event-bus/default'}]}
                : {Pipes: [{Name: 'orders-pipe', CurrentState: 'RUNNING', Source: SRC, Target: DST}], NextToken: 'n'}
        })

        const pipes = await adapter.list()
        expect(pipes.map((pipe) => [pipe.name, pipe.status, pipe.metadata.sourceName, pipe.metadata.targetName])).toEqual([
            ['orders-pipe', 'RUNNING', 'pipe-src (SQS queue)', 'worker (Lambda function)'],
            ['clicks', 'STOPPED', 'clicks (Kinesis stream)', 'default (EventBridge event bus)'],
        ])
        expect((await adapter.list({search: 'kinesis'})).map((pipe) => pipe.name)).toEqual(['clicks'])
    })

    test('describes the filter, batch size, template and role', async () => {
        const {adapter} = stubPipes(() => described)
        expect((await adapter.get('orders-pipe'))?.metadata).toMatchObject({
            filterPatterns: ['{"body":{"kind":["order"]}}'],
            batchSize: 5,
            inputTemplate: '{"id": <$.messageId>}',
            roleArn: ROLE,
        })
    })

    test('a missing pipe is null', async () => {
        const {adapter} = stubPipes(() => { throw Object.assign(new Error('gone'), {name: 'NotFoundException'}) })
        expect(await adapter.get('nope')).toBeNull()
    })

    test('creates an SQS pipe with its filter, batch size and template', async () => {
        const {adapter, sent} = stubPipes((command) => command instanceof DescribePipeCommand ? described : {})

        await adapter.create({values: {name: 'orders-pipe', source: SRC, target: DST, roleArn: ROLE, batchSize: '5', filterPattern: '{ "body": { "kind": ["order"] } }', inputTemplate: '{"id": <$.messageId>}'}})
        expect((sent[0] as CreatePipeCommand).input).toEqual({
            Name: 'orders-pipe',
            Source: SRC,
            SourceParameters: {FilterCriteria: {Filters: [{Pattern: '{"body":{"kind":["order"]}}'}]}, SqsQueueParameters: {BatchSize: 5}},
            Target: DST,
            TargetParameters: {InputTemplate: '{"id": <$.messageId>}'},
            RoleArn: ROLE,
            DesiredState: 'RUNNING',
        })
    })

    test('gives stream and MSK sources the starting position they require', async () => {
        const {adapter, sent} = stubPipes((command) => command instanceof DescribePipeCommand ? described : {})

        await adapter.create({values: {name: 'clicks', source: 'arn:aws:kinesis:us-east-1:000000000000:stream/clicks', target: DST, roleArn: ROLE, startingPosition: 'TRIM_HORIZON', desiredState: 'STOPPED'}})
        expect((sent[0] as CreatePipeCommand).input).toMatchObject({SourceParameters: {KinesisStreamParameters: {StartingPosition: 'TRIM_HORIZON'}}, DesiredState: 'STOPPED'})

        await adapter.create({values: {name: 'k', source: 'arn:aws:kafka:us-east-1:000000000000:cluster/events/abc', target: DST, roleArn: ROLE, topicName: 'orders'}})
        expect((sent[2] as CreatePipeCommand).input.SourceParameters).toEqual({ManagedStreamingKafkaParameters: {TopicName: 'orders', StartingPosition: 'LATEST'}})
    })

    test('refuses pipes the runtime could not run, before calling it', async () => {
        const {adapter, sent} = stubPipes(() => ({}))
        const base = {name: 'p', source: SRC, target: DST, roleArn: ROLE}
        await expect(adapter.create({values: {...base, target: 'arn:aws:s3:::bucket'}})).rejects.toThrow('target must be the ARN of a')
        await expect(adapter.create({values: {...base, source: 'arn:aws:sns:us-east-1:000000000000:topic'}})).rejects.toBeInstanceOf(ValidationError)
        await expect(adapter.create({values: {...base, enrichment: 'arn:aws:sqs:us-east-1:000000000000:q'}})).rejects.toBeInstanceOf(ValidationError)
        await expect(adapter.create({values: {...base, filterPattern: '{kind: order}'}})).rejects.toThrow('filterPattern must be JSON')
        await expect(adapter.create({values: {...base, batchSize: '20000'}})).rejects.toBeInstanceOf(ValidationError)
        await expect(adapter.create({values: {...base, source: 'arn:aws:kafka:us-east-1:000000000000:cluster/c/1'}})).rejects.toThrow('topicName is required')
        await expect(adapter.create({values: {...base, roleArn: 'pipe-role'}})).rejects.toBeInstanceOf(ValidationError)
        expect(sent).toHaveLength(0)
    })

    test('starts, stops and deletes by name', async () => {
        const {adapter, sent} = stubPipes(() => ({}))
        await adapter.stop('orders-pipe')
        await adapter.start('orders-pipe')
        await adapter.delete('orders-pipe')
        expect(sent.map((command) => command.constructor)).toEqual([StopPipeCommand, StartPipeCommand, DeletePipeCommand])
        expect(sent.every((command) => (command as StopPipeCommand).input.Name === 'orders-pipe')).toBe(true)
    })

    test('names every kind of source and target ARN', () => {
        expect(describeArn('arn:aws:dynamodb:us-east-1:000000000000:table/orders/stream/2026-10-25T00:00:00.000')).toBe('orders (DynamoDB stream)')
        expect(describeArn('arn:aws:states:us-east-1:000000000000:stateMachine:flow')).toBe('flow (Step Functions state machine)')
        expect(describeArn('arn:aws:sns:us-east-1:000000000000:alerts')).toBe('alerts (SNS topic)')
    })
})